Repository: eclipse-vertx/vert.x

Problem statement:
HttpClientResponse executed on the a different context thread when resumed outside of its context
I am trying to load the body of an HTTP request. However, in the data handler, I need to do stuff that is asynchronous. So I pause the stream at the beginning of the handler, and resume it when I'm done doing my asynchronous stuff.

Pseudocode:

```
each time I receive a buffer
    pause the response
    asynchronously
        do something with the buffer
        resume the response
```

However, it's not working as I expect it to.

```
val channel = Channel<Byte>()

val lock = Object()
val INITIAL = 1
val PAUSED = 2
val READING = 3
val RESUMED = 4

var state = INITIAL

vertxResponse.handler { buffer ->
    synchronized(lock) {
        if (state != INITIAL && state != RESUMED) {
            println("Fail A")
        }

        state = PAUSED
        vertxResponse.pause()
    }

    async(Unconfined) {
        synchronized(lock) {
            if (state != PAUSED) {
                println("Fail B")
            }
            state = READING
        }
        try {
            buffer.bytes.forEach {
                channel.send(it)
            }
        } catch (t: Throwable) {
            channel.close(t)
        }

        synchronized(lock) {
            if (state != READING) {
                println("Fail C")
            }
            state = RESUMED
            vertxResponse.resume()
        }
    }
}

vertxResponse.endHandler {
    channel.close()
}

vertxResponse.exceptionHandler {
    channel.close(it)
}

bodyIterable = object : CancelableAsyncIterable<Byte> {
    suspend override fun iterator(): CancelableAsyncIterator<Byte> = ChannelAsyncIterator(channel)
}
```

Typical log:

```
Fail A
Fail C
```

Am I getting something wrong? Is it possible my handler gets called again before it has time to pause the reading?


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/http/HttpClientRequest.java b/src/main/java/io/vertx/core/http/HttpClientRequest.java
--- a/src/main/java/io/vertx/core/http/HttpClientRequest.java
+++ b/src/main/java/io/vertx/core/http/HttpClientRequest.java
@@ -328,7 +328,7 @@ default boolean reset() {
    * <p/>
    * <ul>
    *   <li>for HTTP/2, this performs send an HTTP/2 reset frame with the specified error {@code code}</li>
-   *   <li>for HTTP/1.x, this closes the connection after the current in-flight requests are ended</li>
+   *   <li>for HTTP/1.x, this closes the connection when the current request is inflight</li>
    * </ul>
    * <p/>
    * When the request has not yet been sent, the request will be aborted and false is returned as indicator.
diff --git a/src/main/java/io/vertx/core/http/HttpConnection.java b/src/main/java/io/vertx/core/http/HttpConnection.java
--- a/src/main/java/io/vertx/core/http/HttpConnection.java
+++ b/src/main/java/io/vertx/core/http/HttpConnection.java
@@ -68,11 +68,11 @@ default HttpConnection setWindowSize(int windowSize) {
   }
 
   /**
-   * Like {@link #goAway(long, int)} with a last stream id {@code 2^31-1}.
+   * Like {@link #goAway(long, int)} with a last stream id {@code -1} which means to disallow any new stream creation.
    */
   @Fluent
   default HttpConnection goAway(long errorCode) {
-    return goAway(errorCode, Integer.MAX_VALUE);
+    return goAway(errorCode, -1);
   }
 
   /**
diff --git a/src/main/java/io/vertx/core/http/impl/ClientConnection.java b/src/main/java/io/vertx/core/http/impl/ClientConnection.java
deleted file mode 100644
--- a/src/main/java/io/vertx/core/http/impl/ClientConnection.java
+++ /dev/null
@@ -1,623 +0,0 @@
-/*
- * Copyright (c) 2011-2013 The original author or authors
- * ------------------------------------------------------
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
- *
- *     The Eclipse Public License is available at
- *     http://www.eclipse.org/legal/epl-v10.html
- *
- *     The Apache License v2.0 is available at
- *     http://www.opensource.org/licenses/apache2.0.php
- *
- * You may elect to redistribute this code under either of these licenses.
- */
-
-package io.vertx.core.http.impl;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.*;
-import io.netty.handler.codec.http.*;
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.websocketx.*;
-import io.netty.util.ReferenceCountUtil;
-import io.vertx.core.Handler;
-import io.vertx.core.MultiMap;
-import io.vertx.core.VertxException;
-import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.*;
-import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.HttpVersion;
-import io.vertx.core.http.impl.ws.WebSocketFrameInternal;
-import io.vertx.core.impl.ContextImpl;
-import io.vertx.core.logging.Logger;
-import io.vertx.core.logging.LoggerFactory;
-import io.vertx.core.net.NetSocket;
-import io.vertx.core.net.impl.NetSocketImpl;
-import io.vertx.core.net.impl.VertxNetHandler;
-import io.vertx.core.spi.metrics.HttpClientMetrics;
-
-import java.net.URI;
-import java.util.ArrayDeque;
-import java.util.Deque;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Queue;
-
-import static io.vertx.core.http.HttpHeaders.ACCEPT_ENCODING;
-import static io.vertx.core.http.HttpHeaders.CLOSE;
-import static io.vertx.core.http.HttpHeaders.CONNECTION;
-import static io.vertx.core.http.HttpHeaders.DEFLATE_GZIP;
-import static io.vertx.core.http.HttpHeaders.HOST;
-import static io.vertx.core.http.HttpHeaders.KEEP_ALIVE;
-import static io.vertx.core.http.HttpHeaders.TRANSFER_ENCODING;
-
-/**
- *
- * This class is optimised for performance when used on the same event loop. However it can be used safely from other threads.
- *
- * The internal state is protected using the synchronized keyword. If always used on the same event loop, then
- * we benefit from biased locking which makes the overhead of synchronized near zero.
- *
- * @author <a href="http://tfox.org">Tim Fox</a>
- */
-class ClientConnection extends Http1xConnectionBase implements HttpClientConnection, HttpClientStream {
-
-  private static final Logger log = LoggerFactory.getLogger(ClientConnection.class);
-
-  private final HttpClientImpl client;
-  private final boolean ssl;
-  private final String host;
-  private final int port;
-  private final Http1xPool pool;
-  private final Object endpointMetric;
-  // Requests can be pipelined so we need a queue to keep track of requests
-  private final Deque<HttpClientRequestImpl> requests = new ArrayDeque<>();
-  private final HttpClientMetrics metrics;
-  private final HttpVersion version;
-
-  private WebSocketClientHandshaker handshaker;
-  private HttpClientRequestImpl currentRequest;
-  private HttpClientResponseImpl currentResponse;
-  private HttpClientRequestImpl requestForResponse;
-  private WebSocketImpl ws;
-
-  private boolean reset;
-  private boolean paused;
-  private Buffer pausedChunk;
-
-  ClientConnection(HttpVersion version, HttpClientImpl client, Object endpointMetric, ChannelHandlerContext channel, boolean ssl, String host,
-                   int port, ContextImpl context, Http1xPool pool, HttpClientMetrics metrics) {
-    super(client.getVertx(), channel, context);
-    this.client = client;
-    this.ssl = ssl;
-    this.host = host;
-    this.port = port;
-    this.pool = pool;
-    this.metrics = metrics;
-    this.version = version;
-    this.endpointMetric = endpointMetric;
-  }
-
-  public HttpClientMetrics metrics() {
-    return metrics;
-  }
-
-  synchronized HttpClientRequestImpl getCurrentRequest() {
-    return currentRequest;
-  }
-
-  synchronized void toWebSocket(String requestURI, MultiMap headers, WebsocketVersion vers, String subProtocols,
-                   int maxWebSocketFrameSize, Handler<WebSocket> wsConnect) {
-    if (ws != null) {
-      throw new IllegalStateException("Already websocket");
-    }
-
-    try {
-      URI wsuri = new URI(requestURI);
-      if (!wsuri.isAbsolute()) {
-        // Netty requires an absolute url
-        wsuri = new URI((ssl ? "https:" : "http:") + "//" + host + ":" + port + requestURI);
-      }
-      WebSocketVersion version =
-         WebSocketVersion.valueOf((vers == null ?
-           WebSocketVersion.V13 : vers).toString());
-      HttpHeaders nettyHeaders;
-      if (headers != null) {
-        nettyHeaders = new DefaultHttpHeaders();
-        for (Map.Entry<String, String> entry: headers) {
-          nettyHeaders.add(entry.getKey(), entry.getValue());
-        }
-      } else {
-        nettyHeaders = null;
-      }
-      handshaker = WebSocketClientHandshakerFactory.newHandshaker(wsuri, version, subProtocols, false,
-                                                                  nettyHeaders, maxWebSocketFrameSize,!client.getOptions().isSendUnmaskedFrames(),false);
-      ChannelPipeline p = chctx.pipeline();
-      p.addBefore("handler", "handshakeCompleter", new HandshakeInboundHandler(wsConnect, version != WebSocketVersion.V00));
-      handshaker.handshake(chctx.channel()).addListener(future -> {
-        Handler<Throwable> handler = exceptionHandler();
-        if (!future.isSuccess() && handler != null) {
-          handler.handle(future.cause());
-        }
-      });
-    } catch (Exception e) {
-      handleException(e);
-    }
-  }
-
-  private final class HandshakeInboundHandler extends ChannelInboundHandlerAdapter {
-
-    private final boolean supportsContinuation;
-    private final Handler<WebSocket> wsConnect;
-    private final ContextImpl context;
-    private final Queue<Object> buffered = new ArrayDeque<>();
-    private FullHttpResponse response;
-    private boolean handshaking = true;
-
-    public HandshakeInboundHandler(Handler<WebSocket> wsConnect, boolean supportsContinuation) {
-      this.supportsContinuation = supportsContinuation;
-      this.wsConnect = wsConnect;
-      this.context = vertx.getContext();
-    }
-
-    @Override
-    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-      super.channelInactive(ctx);
-      // if still handshaking this means we not got any response back from the server and so need to notify the client
-      // about it as otherwise the client would never been notified.
-      if (handshaking) {
-        handleException(new WebSocketHandshakeException("Connection closed while handshake in process"));
-      }
-    }
-
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-      if (handshaker != null && handshaking) {
-        if (msg instanceof HttpResponse) {
-          HttpResponse resp = (HttpResponse) msg;
-          HttpResponseStatus status = resp.status();
-          if (status.code() != 101) {
-            handshaker = null;
-            close();
-            handleException(new WebsocketRejectedException(status.code()));
-            return;
-          }
-          response = new DefaultFullHttpResponse(resp.protocolVersion(), status);
-          response.headers().add(resp.headers());
-        }
-
-        if (msg instanceof HttpContent) {
-          if (response != null) {
-            response.content().writeBytes(((HttpContent) msg).content());
-            if (msg instanceof LastHttpContent) {
-              response.trailingHeaders().add(((LastHttpContent) msg).trailingHeaders());
-              try {
-                handshakeComplete(ctx, response);
-                chctx.pipeline().remove(HandshakeInboundHandler.this);
-                for (; ; ) {
-                  Object m = buffered.poll();
-                  if (m == null) {
-                    break;
-                  }
-                  ctx.fireChannelRead(m);
-                }
-              } catch (WebSocketHandshakeException e) {
-                close();
-                handleException(e);
-              }
-            }
-          }
-        }
-      } else {
-        buffered.add(msg);
-      }
-    }
-
-    private void handleException(Exception e) {
-      handshaking = false;
-      buffered.clear();
-      Handler<Throwable> handler = exceptionHandler();
-      if (handler != null) {
-        context.executeFromIO(() -> {
-          handler.handle(e);
-        });
-      } else {
-        log.error("Error in websocket handshake", e);
-      }
-    }
-
-    private void handshakeComplete(ChannelHandlerContext ctx, FullHttpResponse response) {
-      handshaking = false;
-      ChannelHandler handler = ctx.pipeline().get(HttpContentDecompressor.class);
-      if (handler != null) {
-        // remove decompressor as its not needed anymore once connection was upgraded to websockets
-        ctx.pipeline().remove(handler);
-      }
-      // Need to set context before constructor is called as writehandler registration needs this
-      ContextImpl.setContext(context);
-      WebSocketImpl webSocket = new WebSocketImpl(vertx, ClientConnection.this, supportsContinuation,
-                                                  client.getOptions().getMaxWebsocketFrameSize(),
-                                                  client.getOptions().getMaxWebsocketMessageSize());
-      ws = webSocket;
-      handshaker.finishHandshake(chctx.channel(), response);
-      ws.subProtocol(handshaker.actualSubprotocol());
-      context.executeFromIO(() -> {
-        log.debug("WebSocket handshake complete");
-        if (metrics != null ) {
-          webSocket.setMetric(metrics.connected(endpointMetric, metric(), webSocket));
-        }
-        wsConnect.handle(webSocket);
-      });
-    }
-  }
-
-  public boolean isValid() {
-    return !reset && chctx.channel().isOpen();
-  }
-
-  int getOutstandingRequestCount() {
-    return requests.size();
-  }
-
-  @Override
-  public void checkDrained() {
-    handleInterestedOpsChanged();
-  }
-
-  @Override
-  public synchronized void handleInterestedOpsChanged() {
-    if (!isNotWritable()) {
-      if (currentRequest != null) {
-        currentRequest.handleDrained();
-      } else if (ws != null) {
-        ws.writable();
-      }
-    }
-  }
-
-  void handleResponse(HttpResponse resp) {
-    if (resp.status().code() == 100) {
-      //If we get a 100 continue it will be followed by the real response later, so we don't remove it yet
-      requestForResponse = requests.peek();
-    } else {
-      requestForResponse = requests.poll();
-    }
-    if (requestForResponse == null) {
-      throw new IllegalStateException("No response handler");
-    }
-    io.netty.handler.codec.http.HttpVersion nettyVersion = resp.protocolVersion();
-    HttpVersion vertxVersion;
-    if (nettyVersion == io.netty.handler.codec.http.HttpVersion.HTTP_1_0) {
-      vertxVersion = HttpVersion.HTTP_1_0;
-    } else if (nettyVersion == io.netty.handler.codec.http.HttpVersion.HTTP_1_1) {
-      vertxVersion = HttpVersion.HTTP_1_1;
-    } else {
-      vertxVersion = null;
-    }
-    HttpClientResponseImpl nResp = new HttpClientResponseImpl(requestForResponse, vertxVersion, this, resp.status().code(), resp.status().reasonPhrase(), new HeadersAdaptor(resp.headers()));
-    currentResponse = nResp;
-    if (metrics != null) {
-      metrics.responseBegin(requestForResponse.metric(), nResp);
-    }
-    if (vertxVersion != null) {
-      requestForResponse.handleResponse(nResp);
-    } else {
-      requestForResponse.handleException(new IllegalStateException("Unsupported HTTP version: " + nettyVersion));
-    }
-  }
-
-  public void doPause() {
-    super.doPause();
-    paused = true;
-  }
-
-  public void doResume() {
-    super.doResume();
-    paused = false;
-    if (pausedChunk != null) {
-      vertx.runOnContext(v -> {
-        if (pausedChunk != null) {
-          Buffer chunk = pausedChunk;
-          pausedChunk = null;
-          currentResponse.handleChunk(chunk);
-        }
-      });
-    }
-  }
-
-  void handleResponseChunk(Buffer buff) {
-    if (paused) {
-      if (pausedChunk == null) {
-        pausedChunk = buff.copy();
-      } else {
-        pausedChunk.appendBuffer(buff);
-      }
-    } else {
-      if (pausedChunk != null) {
-        buff = pausedChunk.appendBuffer(buff);
-        pausedChunk = null;
-      }
-      currentResponse.handleChunk(buff);
-    }
-  }
-
-  void handleResponseEnd(LastHttpContent trailer) {
-    if (metrics != null) {
-      HttpClientRequestBase req = currentResponse.request();
-      Object reqMetric = req.metric();
-      if (req.exceptionOccurred != null) {
-        metrics.requestReset(reqMetric);
-      } else {
-        metrics.responseEnd(reqMetric, currentResponse);
-      }
-    }
-    Buffer last = pausedChunk;
-    pausedChunk = null;
-    currentResponse.handleEnd(last, new HeadersAdaptor(trailer.trailingHeaders()));
-
-    // We don't signal response end for a 100-continue response as a real response will follow
-    // Also we keep the connection open for an HTTP CONNECT
-    if (currentResponse.statusCode() != 100 && requestForResponse.method() != io.vertx.core.http.HttpMethod.CONNECT) {
-
-      boolean close = false;
-      // See https://tools.ietf.org/html/rfc7230#section-6.3
-      String responseConnectionHeader = currentResponse.getHeader(HttpHeaders.Names.CONNECTION);
-      io.vertx.core.http.HttpVersion protocolVersion = client.getOptions().getProtocolVersion();
-      String requestConnectionHeader = requestForResponse.headers().get(HttpHeaders.Names.CONNECTION);
-      // We don't need to protect against concurrent changes on forceClose as it only goes from false -> true
-      if (HttpHeaders.Values.CLOSE.equalsIgnoreCase(responseConnectionHeader) || HttpHeaders.Values.CLOSE.equalsIgnoreCase(requestConnectionHeader)) {
-        // In all cases, if we have a close connection option then we SHOULD NOT treat the connection as persistent
-        close = true;
-      } else if (protocolVersion == io.vertx.core.http.HttpVersion.HTTP_1_0 && !HttpHeaders.Values.KEEP_ALIVE.equalsIgnoreCase(responseConnectionHeader)) {
-        // In the HTTP/1.0 case both request/response need a keep-alive connection header the connection to be persistent
-        // currently Vertx forces the Connection header if keepalive is enabled for 1.0
-        close = true;
-      }
-
-      if (close) {
-        pool.responseEnded(this, true);
-      } else {
-        if (reset) {
-          if (requests.isEmpty()) {
-            pool.responseEnded(this, true);
-          }
-        } else {
-          pool.responseEnded(this, false);
-        }
-      }
-    }
-    requestForResponse = null;
-    currentResponse = null;
-  }
-
-  synchronized void handleWsFrame(WebSocketFrameInternal frame) {
-    if (ws != null) {
-      ws.handleFrame(frame);
-    }
-  }
-
-  protected synchronized void handleClosed() {
-    super.handleClosed();
-    if (ws != null) {
-      ws.handleClosed();
-    }
-    Exception e = new VertxException("Connection was closed");
-
-    // Signal requests failed
-    if (metrics != null) {
-      for (HttpClientRequestImpl req: requests) {
-        metrics.requestReset(req.metric());
-      }
-      if (currentResponse != null) {
-        metrics.requestReset(currentResponse.request().metric());
-      }
-    }
-
-    // Connection was closed - call exception handlers for any requests in the pipeline or one being currently written
-    for (HttpClientRequestImpl req: requests) {
-      if (req != currentRequest) {
-        req.handleException(e);
-      }
-    }
-    if (currentRequest != null) {
-      currentRequest.handleException(e);
-    } else if (currentResponse != null) {
-      currentResponse.handleException(e);
-    }
-  }
-
-  public ContextImpl getContext() {
-    return super.getContext();
-  }
-
-  @Override
-  public void reset(long code) {
-    if (!reset) {
-      reset = true;
-      if (currentRequest != null) {
-        requests.removeLast();
-      }
-      if (requests.size() == 0) {
-        pool.responseEnded(this, true);
-      }
-    }
-  }
-
-  private HttpRequest createRequest(HttpVersion version, HttpMethod method, String rawMethod, String uri, MultiMap headers) {
-    DefaultHttpRequest request = new DefaultHttpRequest(HttpUtils.toNettyHttpVersion(version), HttpUtils.toNettyHttpMethod(method, rawMethod), uri, false);
-    if (headers != null) {
-      for (Map.Entry<String, String> header : headers) {
-        // Todo : multi valued headers
-        request.headers().add(header.getKey(), header.getValue());
-      }
-    }
-    return request;
-  }
-
-  private void prepareHeaders(HttpRequest request, String hostHeader, boolean chunked) {
-    HttpHeaders headers = request.headers();
-    headers.remove(TRANSFER_ENCODING);
-    if (!headers.contains(HOST)) {
-      request.headers().set(HOST, hostHeader);
-    }
-    if (chunked) {
-      HttpHeaders.setTransferEncodingChunked(request);
-    }
-    if (client.getOptions().isTryUseCompression() && request.headers().get(ACCEPT_ENCODING) == null) {
-      // if compression should be used but nothing is specified by the user support deflate and gzip.
-      request.headers().set(ACCEPT_ENCODING, DEFLATE_GZIP);
-    }
-    if (!client.getOptions().isKeepAlive() && client.getOptions().getProtocolVersion() == io.vertx.core.http.HttpVersion.HTTP_1_1) {
-      request.headers().set(CONNECTION, CLOSE);
-    } else if (client.getOptions().isKeepAlive() && client.getOptions().getProtocolVersion() == io.vertx.core.http.HttpVersion.HTTP_1_0) {
-      request.headers().set(CONNECTION, KEEP_ALIVE);
-    }
-  }
-
-  public void writeHead(HttpMethod method, String rawMethod, String uri, MultiMap headers, String hostHeader, boolean chunked) {
-    HttpRequest request = createRequest(version, method, rawMethod, uri, headers);
-    prepareHeaders(request, hostHeader, chunked);
-    writeToChannel(request);
-  }
-
-  public void writeHeadWithContent(HttpMethod method, String rawMethod, String uri, MultiMap headers, String hostHeader, boolean chunked, ByteBuf buf, boolean end) {
-    HttpRequest request = createRequest(version, method, rawMethod, uri, headers);
-    prepareHeaders(request, hostHeader, chunked);
-    if (end) {
-      if (buf != null) {
-        writeToChannel(new AssembledFullHttpRequest(request, buf));
-      } else {
-        writeToChannel(new AssembledFullHttpRequest(request));
-      }
-    } else {
-      writeToChannel(new AssembledHttpRequest(request, buf));
-    }
-  }
-
-  @Override
-  public void writeBuffer(ByteBuf buff, boolean end) {
-    if (end) {
-      if (buff != null && buff.isReadable()) {
-        writeToChannel(new DefaultLastHttpContent(buff, false));
-      } else {
-        writeToChannel(LastHttpContent.EMPTY_LAST_CONTENT);
-      }
-    } else if (buff != null) {
-      writeToChannel(new DefaultHttpContent(buff));
-    }
-  }
-
-  @Override
-  public void writeFrame(int type, int flags, ByteBuf payload) {
-    throw new IllegalStateException("Cannot write an HTTP/2 frame over an HTTP/1.x connection");
-  }
-
-  @Override
-  protected synchronized void handleException(Throwable e) {
-    super.handleException(e);
-    if (currentRequest != null) {
-      currentRequest.handleException(e);
-    } else {
-      HttpClientRequestImpl req = requests.poll();
-      if (req != null) {
-        req.handleException(e);
-      } else if (currentResponse != null) {
-        currentResponse.handleException(e);
-      }
-    }
-  }
-
-  public synchronized void beginRequest(HttpClientRequestImpl req) {
-    if (currentRequest != null) {
-      throw new IllegalStateException("Connection is already writing a request");
-    }
-    if (metrics != null) {
-      Object reqMetric = metrics.requestBegin(endpointMetric, metric(), localAddress(), remoteAddress(), req);
-      req.metric(reqMetric);
-    }
-    this.currentRequest = req;
-    this.requests.add(req);
-  }
-
-  public synchronized void endRequest() {
-    if (currentRequest == null) {
-      throw new IllegalStateException("No write in progress");
-    }
-    if (metrics != null) {
-      metrics.requestEnd(currentRequest.metric());
-    }
-    currentRequest = null;
-    pool.requestEnded(this);
-  }
-
-  @Override
-  public synchronized void close() {
-    if (handshaker == null) {
-      super.close();
-    } else {
-      // make sure everything is flushed out on close
-      endReadAndFlush();
-      // close the websocket connection by sending a close frame.
-      handshaker.close(chctx.channel(), new CloseWebSocketFrame(1000, null));
-    }
-  }
-
-  public NetSocket createNetSocket() {
-    // connection was upgraded to raw TCP socket
-    NetSocketImpl socket = new NetSocketImpl(vertx, chctx, context, client.getSslHelper(), metrics);
-    socket.metric(metric());
-    Map<Channel, NetSocketImpl> connectionMap = new HashMap<>(1);
-    connectionMap.put(chctx.channel(), socket);
-
-    // Flush out all pending data
-    endReadAndFlush();
-
-    // remove old http handlers and replace the old handler with one that handle plain sockets
-    ChannelPipeline pipeline = chctx.pipeline();
-    ChannelHandler inflater = pipeline.get(HttpContentDecompressor.class);
-    if (inflater != null) {
-      pipeline.remove(inflater);
-    }
-    pipeline.remove("codec");
-    pipeline.replace("handler", "handler",  new VertxNetHandler(socket) {
-      @Override
-      public void channelRead(ChannelHandlerContext chctx, Object msg) throws Exception {
-        if (msg instanceof HttpContent) {
-          if (msg instanceof LastHttpContent) {
-            handleResponseEnd((LastHttpContent) msg);
-          }
-          ReferenceCountUtil.release(msg);
-          return;
-        }
-        super.channelRead(chctx, msg);
-      }
-      @Override
-      protected void handleMessage(NetSocketImpl connection, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
-        ByteBuf buf = (ByteBuf) msg;
-        connection.handleMessageReceived(buf);
-      }
-    }.removeHandler(sock -> {
-      pool.removeChannel(chctx.channel());
-    }));
-    return socket;
-  }
-
-  @Override
-  public HttpClientConnection connection() {
-    return this;
-  }
-
-  @Override
-  public HttpVersion version() {
-    // Used to determine the http version in the HttpClientRequest#sendHead handler , for HTTP/1.1 it will
-    // not yet know but it will for HTTP/2
-    return null;
-  }
-
-  @Override
-  public int id() {
-    return -1;
-  }
-}
diff --git a/src/main/java/io/vertx/core/http/impl/ConnectionManager.java b/src/main/java/io/vertx/core/http/impl/ConnectionManager.java
--- a/src/main/java/io/vertx/core/http/impl/ConnectionManager.java
+++ b/src/main/java/io/vertx/core/http/impl/ConnectionManager.java
@@ -16,93 +16,55 @@
 
 package io.vertx.core.http.impl;
 
-import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.channel.ChannelPipeline;
-import io.netty.handler.codec.http.DefaultFullHttpRequest;
-import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.HttpClientCodec;
-import io.netty.handler.codec.http.HttpClientUpgradeHandler;
-import io.netty.handler.codec.http.HttpContentDecompressor;
-import io.netty.handler.codec.http.HttpHeaderNames;
-import io.netty.handler.codec.http.HttpMethod;
-import io.netty.handler.codec.http.LastHttpContent;
-import io.netty.handler.codec.http2.Http2Exception;
-import io.netty.handler.logging.LoggingHandler;
-import io.netty.handler.ssl.SslHandler;
-import io.netty.handler.timeout.IdleStateHandler;
-import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
-import io.vertx.core.http.ConnectionPoolTooBusyException;
-import io.vertx.core.http.HttpClientOptions;
+import io.vertx.core.http.HttpConnection;
 import io.vertx.core.http.HttpVersion;
-import io.vertx.core.impl.ContextImpl;
-import io.vertx.core.impl.VertxInternal;
-import io.vertx.core.logging.Logger;
-import io.vertx.core.logging.LoggerFactory;
-import io.vertx.core.net.ProxyType;
-import io.vertx.core.net.SocketAddress;
-import io.vertx.core.net.impl.ChannelProvider;
-import io.vertx.core.net.impl.ProxyChannelProvider;
-import io.vertx.core.net.impl.SSLHelper;
+import io.vertx.core.http.impl.pool.Pool;
+import io.vertx.core.http.impl.pool.Waiter;
+import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
 
-import javax.net.ssl.SSLHandshakeException;
-import java.util.ArrayDeque;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Queue;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
 
 /**
+ * The connection manager associates remote hosts with pools, it also tracks all connections so they can be closed
+ * when the manager is closed.
  *
  * @author <a href="http://tfox.org">Tim Fox</a>
  */
-public class ConnectionManager {
+class ConnectionManager {
 
-  static final Logger log = LoggerFactory.getLogger(ConnectionManager.class);
-
-  private final QueueManager wsQM = new QueueManager(); // The queue manager for websockets
-  private final QueueManager requestQM = new QueueManager(); // The queue manager for requests
-  private final VertxInternal vertx;
-  private final SSLHelper sslHelper;
-  private final HttpClientOptions options;
-  private final HttpClientImpl client;
-  private final boolean keepAlive;
-  private final boolean pipelining;
   private final int maxWaitQueueSize;
-  private final int http2MaxConcurrency;
-  private final boolean logEnabled;
-  private final ChannelConnector connector;
-  private final HttpClientMetrics metrics;
-
-  ConnectionManager(HttpClientImpl client, HttpClientMetrics metrics) {
+  private final HttpClientMetrics metrics; // Shall be removed later combining the PoolMetrics with HttpClientMetrics
+  private final HttpClientImpl client;
+  private final Map<Channel, HttpClientConnection> connectionMap = new ConcurrentHashMap<>();
+  private final Map<EndpointKey, Endpoint> endpointMap = new ConcurrentHashMap<>();
+  private final HttpVersion version;
+  private final long maxSize;
+
+  ConnectionManager(HttpClientImpl client,
+                    HttpClientMetrics metrics,
+                    HttpVersion version,
+                    long maxSize,
+                    int maxWaitQueueSize) {
     this.client = client;
-    this.sslHelper = client.getSslHelper();
-    this.options = client.getOptions();
-    this.vertx = client.getVertx();
-    this.keepAlive = client.getOptions().isKeepAlive();
-    this.pipelining = client.getOptions().isPipelining();
-    this.maxWaitQueueSize = client.getOptions().getMaxWaitQueueSize();
-    this.http2MaxConcurrency = options.getHttp2MultiplexingLimit() < 1 ? Integer.MAX_VALUE : options.getHttp2MultiplexingLimit();
-    this.logEnabled = client.getOptions().getLogActivity();
-    this.connector = new ChannelConnector();
+    this.maxWaitQueueSize = maxWaitQueueSize;
     this.metrics = metrics;
+    this.maxSize = maxSize;
+    this.version = version;
   }
 
-  HttpClientMetrics metrics() {
-    return metrics;
-  }
-
-  static final class ConnectionKey {
+  private static final class EndpointKey {
 
     private final boolean ssl;
     private final int port;
     private final String host;
 
-    public ConnectionKey(boolean ssl, int port, String host) {
+    EndpointKey(boolean ssl, int port, String host) {
       this.ssl = ssl;
       this.host = host;
       this.port = port;
@@ -113,7 +75,7 @@ public boolean equals(Object o) {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
 
-      ConnectionKey that = (ConnectionKey) o;
+      EndpointKey that = (EndpointKey) o;
 
       if (ssl != that.ssl) return false;
       if (port != that.port) return false;
@@ -131,434 +93,77 @@ public int hashCode() {
     }
   }
 
-  /**
-   * The queue manager manages the connection queues for a given usage, the idea is to split
-   * queues for HTTP requests and websockets. A websocket uses a pool of connections
-   * usually ugpraded from HTTP/1.1, HTTP requests may ask for HTTP/2 connections but obtain
-   * only HTTP/1.1 connections.
-   */
-  private class QueueManager {
-
-    private final Map<Channel, HttpClientConnection> connectionMap = new ConcurrentHashMap<>();
-    private final Map<ConnectionKey, ConnQueue> queueMap = new ConcurrentHashMap<>();
+  class Endpoint {
 
-    ConnQueue getConnQueue(String peerHost, boolean ssl, int port, String host, HttpVersion version) {
-      ConnectionKey key = new ConnectionKey(ssl, port, peerHost);
-      return queueMap.computeIfAbsent(key, targetAddress -> new ConnQueue(version, this, peerHost, host, port, ssl, key));
-    }
+    private final Pool<HttpClientConnection> pool;
+    private final Object metric;
 
-    public void close() {
-      for (ConnQueue queue: queueMap.values()) {
-        queue.closeAllConnections();
-      }
-      queueMap.clear();
-      for (HttpClientConnection conn : connectionMap.values()) {
-        conn.close();
-      }
+    public Endpoint(Pool<HttpClientConnection> pool, Object metric) {
+      this.pool = pool;
+      this.metric = metric;
     }
   }
 
-  public void getConnectionForWebsocket(boolean ssl, int port, String host, Waiter waiter) {
-    ConnQueue connQueue = wsQM.getConnQueue(host, ssl, port, host, HttpVersion.HTTP_1_1);
-    connQueue.getConnection(waiter);
-  }
-
-  public void getConnectionForRequest(HttpVersion version, String peerHost, boolean ssl, int port, String host, Waiter waiter) {
-    if (!keepAlive && pipelining) {
-      waiter.handleFailure(new IllegalStateException("Cannot have pipelining with no keep alive"));
-    } else {
-      ConnQueue connQueue = requestQM.getConnQueue(peerHost, ssl, port, host, version);
-      connQueue.getConnection(waiter);
-    }
-  }
-
-  public void close() {
-    wsQM.close();
-    requestQM.close();
-    if (metrics != null) {
-      metrics.close();
-    }
-  }
-
-  /**
-   * The connection queue delegates to the connection pool, the pooling strategy.
-   *
-   * - HTTP/1.x pools several connections
-   * - HTTP/2 uses a single connection
-   *
-   * After a queue is initialized with an HTTP/2 pool, this pool changed to an HTTP/1/1
-   * pool if the server does not support HTTP/2 or after negotiation. In this situation
-   * all waiters on this queue will use HTTP/1.1 connections.
-   */
-  public class ConnQueue {
-
-    private final QueueManager mgr;
-    private final String peerHost;
-    private final boolean ssl;
-    private final int port;
-    private final String host;
-    private final ConnectionKey key;
-    private final Queue<Waiter> waiters = new ArrayDeque<>();
-    private Pool<HttpClientConnection> pool;
-    private int connCount;
-    private final int maxSize;
-    final Object metric;
-
-    ConnQueue(HttpVersion version, QueueManager mgr, String peerHost, String host, int port, boolean ssl, ConnectionKey key) {
-      this.key = key;
-      this.host = host;
-      this.port = port;
-      this.ssl = ssl;
-      this.peerHost = peerHost;
-      this.mgr = mgr;
-      if (version == HttpVersion.HTTP_2) {
-        maxSize = options.getHttp2MaxPoolSize();
-        pool =  (Pool)new Http2Pool(this, client, metrics, mgr.connectionMap, http2MaxConcurrency, logEnabled, options.getHttp2MaxPoolSize(), options.getHttp2ConnectionWindowSize());
-      } else {
-        maxSize = options.getMaxPoolSize();
-        pool = (Pool)new Http1xPool(client, metrics, options, this, mgr.connectionMap, version, options.getMaxPoolSize(), host, port);
-      }
-      this.metric = metrics != null ? metrics.createEndpoint(host, port, maxSize) : null;
-    }
-
-    public synchronized void getConnection(Waiter waiter) {
-      HttpClientConnection conn = pool.pollConnection();
-      if (conn != null && conn.isValid()) {
-        ContextImpl context = waiter.context;
-        if (context == null) {
-          context = conn.getContext();
-        } else if (context != conn.getContext()) {
-          ConnectionManager.log.warn("Reusing a connection with a different context: an HttpClient is probably shared between different Verticles");
-        }
-        context.runOnContext(v -> deliverStream(conn, waiter));
-      } else {
-        if (pool.canCreateConnection(connCount)) {
-          // Create a new connection
-          createNewConnection(waiter);
-        } else {
-          // Wait in queue
-          if (maxWaitQueueSize < 0 || waiters.size() < maxWaitQueueSize) {
+  void getConnection(String peerHost, boolean ssl, int port, String host,
+                     Handler<HttpConnection> connectionHandler,
+                     BiFunction<ContextInternal, HttpClientConnection, Boolean> onSuccess,
+                     BiConsumer<ContextInternal, Throwable> onFailure) {
+    EndpointKey key = new EndpointKey(ssl, port, peerHost);
+    while (true) {
+      Endpoint endpoint = endpointMap.computeIfAbsent(key, targetAddress -> {
+        int maxPoolSize = Math.max(client.getOptions().getMaxPoolSize(), client.getOptions().getHttp2MaxPoolSize());
+        Object metric = metrics != null ? metrics.createEndpoint(host, port, maxPoolSize) : null;
+        HttpChannelConnector connector = new HttpChannelConnector(client, metric, version, ssl, peerHost, host, port);
+        Pool<HttpClientConnection> pool = new Pool<>(connector, maxWaitQueueSize, maxSize,
+          v -> {
             if (metrics != null) {
-              waiter.metric = metrics.enqueueRequest(metric);
+              metrics.closeEndpoint(host, port, metric);
             }
-            waiters.add(waiter);
-          } else {
-            waiter.handleFailure(new ConnectionPoolTooBusyException("Connection pool reached max wait queue size of " + maxWaitQueueSize));
-          }
-        }
-      }
-    }
-
-    /**
-     * Handle the connection if the waiter is not cancelled, otherwise recycle the connection.
-     *
-     * @param conn the connection
-     */
-    void deliverStream(HttpClientConnection conn, Waiter waiter) {
-      if (!conn.isValid()) {
-        // The connection has been closed - closed connections can be in the pool
-        // Get another connection - Note that we DO NOT call connectionClosed() on the pool at this point
-        // that is done asynchronously in the connection closeHandler()
-        getConnection(waiter);
-      } else if (waiter.isCancelled()) {
-        pool.recycle(conn);
-      } else {
-        HttpClientStream stream;
-        try {
-          stream = pool.createStream(conn);
-        } catch (Exception e) {
-          getConnection(waiter);
-          return;
-        }
-        waiter.handleStream(stream);
-      }
-    }
-
-    void closeAllConnections() {
-      pool.closeAllConnections();
-    }
-
-    private void createNewConnection(Waiter waiter) {
-      connCount++;
-      ContextImpl context;
-      if (waiter.context == null) {
-        // Embedded
-        context = vertx.getOrCreateContext();
-      } else {
-        context = waiter.context;
-      }
-      sslHelper.validate(vertx);
-      Bootstrap bootstrap = new Bootstrap();
-      bootstrap.group(context.nettyEventLoop());
-      bootstrap.channel(vertx.transport().channelType(false));
-      connector.connect(this, bootstrap, context, peerHost, ssl, pool.version(), host, port, waiter);
-    }
-
-    /**
-     * @return the next non-canceled waiters in the queue
-     */
-    Waiter getNextWaiter() {
-      Waiter waiter = waiters.poll();
-      if (metrics != null && waiter != null) {
-        metrics.dequeueRequest(metric, waiter.metric);
-      }
-      while (waiter != null && waiter.isCancelled()) {
-        waiter = waiters.poll();
-        if (metrics != null && waiter != null) {
-          metrics.dequeueRequest(metric, waiter.metric);
-        }
-      }
-      return waiter;
-    }
-
-    // Called if the connection is actually closed OR the connection attempt failed
-    public synchronized void connectionClosed() {
-      connCount--;
-      Waiter waiter = getNextWaiter();
-      if (waiter != null) {
-        // There's a waiter - so it can have a new connection
-        createNewConnection(waiter);
-      } else if (connCount == 0) {
-        // No waiters and no connections - remove the ConnQueue
-        mgr.queueMap.remove(key);
-        if (metrics != null) {
-          metrics.closeEndpoint(host, port, metric);
-        }
-      }
-    }
-
-    private void handshakeFailure(ContextImpl context, Channel ch, Throwable cause, Waiter waiter) {
-      SSLHandshakeException sslException = new SSLHandshakeException("Failed to create SSL connection");
-      if (cause != null) {
-        sslException.initCause(cause);
-      }
-      connectionFailed(context, ch, waiter::handleFailure, sslException);
-    }
-
-    private void fallbackToHttp1x(Channel ch, ContextImpl context, HttpVersion fallbackVersion, int port, String host, Waiter waiter) {
-      // change the pool to Http1xPool
-      synchronized (this) {
-        pool = (Pool)new Http1xPool(client, ConnectionManager.this.metrics, options, this, mgr.connectionMap, fallbackVersion, options.getMaxPoolSize(), host, port);
-      }
-      http1xConnected(fallbackVersion, context, port, host, ch, waiter);
-    }
-
-    private void http1xConnected(HttpVersion version, ContextImpl context, int port, String host, Channel ch, Waiter waiter) {
-      ((Http1xPool)(Pool)pool).createConn(context, ch, waiter);
-    }
-
-    private void http2Connected(ContextImpl context, Channel ch, Waiter waiter, boolean upgrade) {
-      context.executeFromIO(() -> {
-        try {
-          ((Http2Pool)(Pool)pool).createConn(context, ch, waiter, upgrade);
-        } catch (Http2Exception e) {
-          connectionFailed(context, ch, waiter::handleFailure, e);
-        }
+            endpointMap.remove(key);
+          },
+          connectionMap::put,
+          connectionMap::remove);
+        return new Endpoint(pool, metric);
       });
-    }
-
-    private void connectionFailed(ContextImpl context, Channel ch, Handler<Throwable> connectionExceptionHandler,
-        Throwable t) {
-      // If no specific exception handler is provided, fall back to the HttpClient's exception handler.
-      // If that doesn't exist just log it
-      Handler<Throwable> exHandler =
-          connectionExceptionHandler == null ? log::error : connectionExceptionHandler;
-
-      context.executeFromIO(() -> {
-        connectionClosed();
-        try {
-          ch.close();
-        } catch (Exception ignore) {
-        }
-        exHandler.handle(t);
-      });
-    }
-  }
-
-  /**
-   * The logic for the connection pool because HTTP/1 and HTTP/2 have different pooling logics.
-   */
-  interface Pool<C extends HttpClientConnection> {
-
-    HttpVersion version();
-
-    C pollConnection();
-
-    /**
-     * Determine when a new connection should be created
-     *
-     * @param connCount the actual connection count including the one being created
-     * @return true whether or not a new connection can be created
-     */
-    boolean canCreateConnection(int connCount);
-
-    void closeAllConnections();
-
-    void recycle(C conn);
-
-    HttpClientStream createStream(C conn) throws Exception;
-
-  }
-
-  /**
-   * The ChannelConnector performs the channel configuration and connection according to the
-   * client options and the protocol version.
-   * When the channel connects or fails to connect, it calls back the ConnQueue that initiated the
-   * connection.
-   */
-  private class ChannelConnector {
-
-    protected void connect(
-        ConnQueue queue,
-        Bootstrap bootstrap,
-        ContextImpl context,
-        String peerHost,
-        boolean ssl,
-        HttpVersion version,
-        String host,
-        int port,
-        Waiter waiter) {
-
-      applyConnectionOptions(options, bootstrap);
-
-      ChannelProvider channelProvider;
-      // http proxy requests are handled in HttpClientImpl, everything else can use netty proxy handler
-      if (options.getProxyOptions() == null || !ssl && options.getProxyOptions().getType()==ProxyType.HTTP ) {
-        channelProvider = ChannelProvider.INSTANCE;
+      Object metric;
+      if (metrics != null) {
+        metric = metrics.enqueueRequest(endpoint.metric);
       } else {
-        channelProvider = ProxyChannelProvider.INSTANCE;
-      }
-
-      boolean useAlpn = options.isUseAlpn();
-      Handler<Channel> channelInitializer = ch -> {
-
-        // Configure pipeline
-        ChannelPipeline pipeline = ch.pipeline();
-        if (ssl) {
-          SslHandler sslHandler = new SslHandler(sslHelper.createEngine(client.getVertx(), peerHost, port, options.isForceSni() ? peerHost : null));
-          ch.pipeline().addLast("ssl", sslHandler);
-          // TCP connected, so now we must do the SSL handshake
-          sslHandler.handshakeFuture().addListener(fut -> {
-            if (fut.isSuccess()) {
-              String protocol = sslHandler.applicationProtocol();
-              if (useAlpn) {
-                if ("h2".equals(protocol)) {
-                  applyHttp2ConnectionOptions(ch.pipeline());
-                  queue.http2Connected(context, ch, waiter, false);
-                } else {
-                  applyHttp1xConnectionOptions(ch.pipeline(), context);
-                  HttpVersion fallbackProtocol = "http/1.0".equals(protocol) ?
-                    HttpVersion.HTTP_1_0 : HttpVersion.HTTP_1_1;
-                  queue.fallbackToHttp1x(ch, context, fallbackProtocol, port, host, waiter);
-                }
-              } else {
-                applyHttp1xConnectionOptions(ch.pipeline(), context);
-                queue.http1xConnected(version, context, port, host, ch, waiter);
-              }
-            } else {
-              queue.handshakeFailure(context, ch, fut.cause(), waiter);
-            }
-          });
-        } else {
-          if (version == HttpVersion.HTTP_2) {
-            if (options.isHttp2ClearTextUpgrade()) {
-              HttpClientCodec httpCodec = new HttpClientCodec();
-              class UpgradeRequestHandler extends ChannelInboundHandlerAdapter {
-                @Override
-                public void channelActive(ChannelHandlerContext ctx) throws Exception {
-                  DefaultFullHttpRequest upgradeRequest =
-                      new DefaultFullHttpRequest(io.netty.handler.codec.http.HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
-                  String hostHeader = peerHost;
-                  if (port != 80) {
-                    hostHeader += ":" + port;
-                  }
-                  upgradeRequest.headers().set(HttpHeaderNames.HOST, hostHeader);
-                  ctx.writeAndFlush(upgradeRequest);
-                  ctx.fireChannelActive();
-                }
-                @Override
-                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-                  if (msg instanceof LastHttpContent) {
-                    ChannelPipeline p = ctx.pipeline();
-                    p.remove(httpCodec);
-                    p.remove(this);
-                    // Upgrade handler will remove itself
-                    applyHttp1xConnectionOptions(ch.pipeline(), context);
-                    queue.fallbackToHttp1x(ch, context, HttpVersion.HTTP_1_1, port, host, waiter);
-                  }
-                }
-                @Override
-                public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
-                  super.userEventTriggered(ctx, evt);
-                  if (evt == HttpClientUpgradeHandler.UpgradeEvent.UPGRADE_SUCCESSFUL) {
-                    ctx.pipeline().remove(this);
-                    // Upgrade handler will remove itself
-                  }
-                }
-              }
-              VertxHttp2ClientUpgradeCodec upgradeCodec = new VertxHttp2ClientUpgradeCodec(client.getOptions().getInitialSettings()) {
-                @Override
-                public void upgradeTo(ChannelHandlerContext ctx, FullHttpResponse upgradeResponse) throws Exception {
-                  applyHttp2ConnectionOptions(pipeline);
-                  queue.http2Connected(context, ch, waiter, true);
-                }
-              };
-              HttpClientUpgradeHandler upgradeHandler = new HttpClientUpgradeHandler(httpCodec, upgradeCodec, 65536);
-              ch.pipeline().addLast(httpCodec, upgradeHandler, new UpgradeRequestHandler());
-            } else {
-              applyHttp2ConnectionOptions(pipeline);
-            }
-          } else {
-            applyHttp1xConnectionOptions(pipeline, context);
+        metric = null;
+      }
+      if (endpoint.pool.getConnection(new Waiter<HttpClientConnection>(client.getVertx().getOrCreateContext()) {
+        @Override
+        public void initConnection(ContextInternal ctx, HttpClientConnection conn) {
+          if (connectionHandler != null) {
+            ctx.executeFromIO(() -> {
+              connectionHandler.handle(conn);
+            });
           }
         }
-      };
-
-      Handler<AsyncResult<Channel>> channelHandler = res -> {
-
-        if (res.succeeded()) {
-          Channel ch = res.result();
-          if (!ssl) {
-            if (ch.pipeline().get(HttpClientUpgradeHandler.class) != null) {
-              // Upgrade handler do nothing
-            } else {
-              if (version == HttpVersion.HTTP_2 && !options.isHttp2ClearTextUpgrade()) {
-                queue.http2Connected(context, ch, waiter, false);
-              } else {
-                queue.http1xConnected(version, context, port, host, ch, waiter);
-              }
-            }
+        @Override
+        public void handleFailure(ContextInternal ctx, Throwable failure) {
+          if (metrics != null) {
+            metrics.dequeueRequest(endpoint.metric, metric);
           }
-        } else {
-          queue.connectionFailed(context, null, waiter::handleFailure, res.cause());
+          onFailure.accept(ctx, failure);
         }
-      };
-
-      channelProvider.connect(vertx, bootstrap, options.getProxyOptions(), SocketAddress.inetSocketAddress(port, host), channelInitializer, channelHandler);
-    }
-
-    void applyConnectionOptions(HttpClientOptions options, Bootstrap bootstrap) {
-      vertx.transport().configure(options, bootstrap);
-    }
-
-    void applyHttp2ConnectionOptions(ChannelPipeline pipeline) {
-      if (options.getIdleTimeout() > 0) {
-        pipeline.addLast("idle", new IdleStateHandler(0, 0, options.getIdleTimeout()));
+        @Override
+        public boolean handleConnection(ContextInternal ctx, HttpClientConnection conn) throws Exception {
+          if (metrics != null) {
+            metrics.dequeueRequest(endpoint.metric, metric);
+          }
+          return onSuccess.apply(ctx, conn);
+        }
+      })) {
+        break;
       }
     }
+  }
 
-    void applyHttp1xConnectionOptions(ChannelPipeline pipeline, ContextImpl context) {
-      if (logEnabled) {
-        pipeline.addLast("logging", new LoggingHandler());
-      }
-      pipeline.addLast("codec", new HttpClientCodec(options.getMaxInitialLineLength(), options.getMaxHeaderSize(),
-              options.getMaxChunkSize(), false, false, options.getDecoderInitialBufferSize()));
-      if (options.isTryUseCompression()) {
-        pipeline.addLast("inflater", new HttpContentDecompressor(true));
-      }
-      if (options.getIdleTimeout() > 0) {
-        pipeline.addLast("idle", new IdleStateHandler(0, 0, options.getIdleTimeout()));
-      }
+  public void close() {
+    endpointMap.clear();
+    for (HttpClientConnection conn : connectionMap.values()) {
+      conn.close();
     }
   }
 }
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
@@ -0,0 +1,735 @@
+/*
+ * Copyright (c) 2011-2013 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ *     The Eclipse Public License is available at
+ *     http://www.eclipse.org/legal/epl-v10.html
+ *
+ *     The Apache License v2.0 is available at
+ *     http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.vertx.core.http.impl;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.*;
+import io.netty.handler.codec.http.*;
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.websocketx.*;
+import io.netty.util.ReferenceCountUtil;
+import io.vertx.core.*;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.*;
+import io.vertx.core.http.HttpMethod;
+import io.vertx.core.http.HttpVersion;
+import io.vertx.core.http.impl.pool.ConnectionListener;
+import io.vertx.core.http.impl.ws.WebSocketFrameInternal;
+import io.vertx.core.impl.ContextImpl;
+import io.vertx.core.logging.Logger;
+import io.vertx.core.logging.LoggerFactory;
+import io.vertx.core.net.NetSocket;
+import io.vertx.core.net.impl.NetSocketImpl;
+import io.vertx.core.net.impl.VertxNetHandler;
+import io.vertx.core.spi.metrics.HttpClientMetrics;
+
+import java.net.URI;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Map;
+import java.util.Queue;
+
+import static io.vertx.core.http.HttpHeaders.ACCEPT_ENCODING;
+import static io.vertx.core.http.HttpHeaders.CLOSE;
+import static io.vertx.core.http.HttpHeaders.CONNECTION;
+import static io.vertx.core.http.HttpHeaders.DEFLATE_GZIP;
+import static io.vertx.core.http.HttpHeaders.HOST;
+import static io.vertx.core.http.HttpHeaders.KEEP_ALIVE;
+import static io.vertx.core.http.HttpHeaders.TRANSFER_ENCODING;
+
+/**
+ *
+ * This class is optimised for performance when used on the same event loop. However it can be used safely from other threads.
+ *
+ * The internal state is protected using the synchronized keyword. If always used on the same event loop, then
+ * we benefit from biased locking which makes the overhead of synchronized near zero.
+ *
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ */
+class Http1xClientConnection extends Http1xConnectionBase implements HttpClientConnection {
+
+  private static final Logger log = LoggerFactory.getLogger(Http1xClientConnection.class);
+
+  private final ConnectionListener<HttpClientConnection> listener;
+  private final HttpClientImpl client;
+  private final boolean ssl;
+  private final String host;
+  private final int port;
+  private final Object endpointMetric;
+  private final HttpClientMetrics metrics;
+  private final HttpVersion version;
+
+  private WebSocketClientHandshaker handshaker;
+  private WebSocketImpl ws;
+
+  private final Deque<StreamImpl> pending = new ArrayDeque<>();
+  private final Deque<StreamImpl> inflight = new ArrayDeque<>();
+  private StreamImpl currentRequest;
+  private StreamImpl currentResponse;
+
+  private boolean paused;
+  private Buffer pausedChunk;
+
+  Http1xClientConnection(ConnectionListener<HttpClientConnection> listener,
+                         HttpVersion version,
+                         HttpClientImpl client,
+                         Object endpointMetric,
+                         ChannelHandlerContext channel,
+                         boolean ssl,
+                         String host,
+                         int port,
+                         ContextImpl context,
+                         HttpClientMetrics metrics) {
+    super(client.getVertx(), channel, context);
+    this.listener = listener;
+    this.client = client;
+    this.ssl = ssl;
+    this.host = host;
+    this.port = port;
+    this.metrics = metrics;
+    this.version = version;
+    this.endpointMetric = endpointMetric;
+  }
+
+  private static class StreamImpl implements HttpClientStream {
+
+    private final Http1xClientConnection conn;
+    private final Handler<AsyncResult<HttpClientStream>> handler;
+    private final HttpClientRequestImpl request;
+    private HttpClientResponseImpl response;
+    private boolean requestEnded;
+    private boolean responseEnded;
+    private boolean reset;
+    private boolean close;
+    private boolean upgraded;
+
+    StreamImpl(Http1xClientConnection conn, HttpClientRequestImpl request, Handler<AsyncResult<HttpClientStream>> handler) {
+      this.request = request;
+      this.conn = conn;
+      this.handler = handler;
+    }
+
+    @Override
+    public int id() {
+      return -1;
+    }
+
+    @Override
+    public HttpVersion version() {
+      return conn.version;
+    }
+
+    @Override
+    public HttpClientConnection connection() {
+      return conn;
+    }
+
+    @Override
+    public Context getContext() {
+      return conn.context;
+    }
+
+    private HttpRequest createRequest(HttpVersion version, HttpMethod method, String rawMethod, String uri, MultiMap headers) {
+      DefaultHttpRequest request = new DefaultHttpRequest(HttpUtils.toNettyHttpVersion(version), HttpUtils.toNettyHttpMethod(method, rawMethod), uri, false);
+      if (headers != null) {
+        for (Map.Entry<String, String> header : headers) {
+          // Todo : multi valued headers
+          request.headers().add(header.getKey(), header.getValue());
+        }
+      }
+      return request;
+    }
+
+    private void prepareHeaders(HttpRequest request, String hostHeader, boolean chunked) {
+      HttpHeaders headers = request.headers();
+      headers.remove(TRANSFER_ENCODING);
+      if (!headers.contains(HOST)) {
+        request.headers().set(HOST, hostHeader);
+      }
+      if (chunked) {
+        HttpUtil.setTransferEncodingChunked(request, true);
+      }
+      if (conn.client.getOptions().isTryUseCompression() && request.headers().get(ACCEPT_ENCODING) == null) {
+        // if compression should be used but nothing is specified by the user support deflate and gzip.
+        request.headers().set(ACCEPT_ENCODING, DEFLATE_GZIP);
+      }
+      if (!conn.client.getOptions().isKeepAlive() && conn.client.getOptions().getProtocolVersion() == io.vertx.core.http.HttpVersion.HTTP_1_1) {
+        request.headers().set(CONNECTION, CLOSE);
+      } else if (conn.client.getOptions().isKeepAlive() && conn.client.getOptions().getProtocolVersion() == io.vertx.core.http.HttpVersion.HTTP_1_0) {
+        request.headers().set(CONNECTION, KEEP_ALIVE);
+      }
+    }
+
+    public void writeHead(HttpMethod method, String rawMethod, String uri, MultiMap headers, String hostHeader, boolean chunked) {
+      HttpRequest request = createRequest(conn.version, method, rawMethod, uri, headers);
+      prepareHeaders(request, hostHeader, chunked);
+      conn.writeToChannel(request);
+    }
+
+    public void writeHeadWithContent(HttpMethod method, String rawMethod, String uri, MultiMap headers, String hostHeader, boolean chunked, ByteBuf buf, boolean end) {
+      HttpRequest request = createRequest(conn.version, method, rawMethod, uri, headers);
+      prepareHeaders(request, hostHeader, chunked);
+      if (end) {
+        if (buf != null) {
+          conn.writeToChannel(new AssembledFullHttpRequest(request, buf));
+        } else {
+          conn.writeToChannel(new AssembledFullHttpRequest(request));
+        }
+      } else {
+        conn.writeToChannel(new AssembledHttpRequest(request, buf));
+      }
+    }
+
+    @Override
+    public void writeBuffer(ByteBuf buff, boolean end) {
+      if (end) {
+        if (buff != null && buff.isReadable()) {
+          conn.writeToChannel(new DefaultLastHttpContent(buff, false));
+        } else {
+          conn.writeToChannel(LastHttpContent.EMPTY_LAST_CONTENT);
+        }
+      } else if (buff != null) {
+        conn.writeToChannel(new DefaultHttpContent(buff));
+      }
+    }
+
+    @Override
+    public void writeFrame(int type, int flags, ByteBuf payload) {
+      throw new IllegalStateException("Cannot write an HTTP/2 frame over an HTTP/1.x connection");
+    }
+
+    @Override
+    public void doSetWriteQueueMaxSize(int size) {
+      conn.doSetWriteQueueMaxSize(size);
+    }
+
+    @Override
+    public boolean isNotWritable() {
+      return conn.isNotWritable();
+    }
+
+    @Override
+    public void checkDrained() {
+      conn.handleInterestedOpsChanged();
+    }
+
+    @Override
+    public void doPause() {
+      conn.doPause();
+    }
+
+    @Override
+    public void doResume() {
+      conn.doResume();
+    }
+
+    @Override
+    public void reset(long code) {
+      synchronized (conn) {
+        if (request == null) {
+          throw new IllegalStateException("Sanity check");
+        }
+        if (!reset) {
+          reset = true;
+          if (!requestEnded || !responseEnded) {
+            conn.close();
+          }
+        }
+      }
+    }
+
+    public void beginRequest() {
+      synchronized (conn) {
+        if (conn.currentRequest != this) {
+          throw new IllegalStateException("Connection is already writing another request");
+        }
+        if (conn.metrics != null) {
+          Object reqMetric = conn.metrics.requestBegin(conn.endpointMetric, conn.metric(), conn.localAddress(), conn.remoteAddress(), request);
+          request.metric(reqMetric);
+        }
+        conn.inflight.add(conn.currentRequest);
+      }
+    }
+
+    public void endRequest() {
+      StreamImpl next;
+      synchronized (conn) {
+        if (conn.currentRequest != this) {
+          throw new IllegalStateException("No write in progress");
+        }
+        if (conn.metrics != null) {
+          conn.metrics.requestEnd(conn.currentRequest.request.metric());
+        }
+        requestEnded = true;
+        // Should take care of pending list ????
+        checkLifecycle();
+        // Check pipelined pending request
+        next = conn.currentRequest = conn.pending.poll();
+        if (next == null) {
+          return;
+        }
+      }
+      // Should trampoline ?
+      next.handler.handle(Future.succeededFuture(next));
+    }
+
+    @Override
+    public NetSocket createNetSocket() {
+      synchronized (conn) {
+        if (responseEnded) {
+          throw new IllegalStateException("Request already ended");
+        }
+        if (upgraded) {
+          throw new IllegalStateException("Request already upgraded to NetSocket");
+        }
+        upgraded = true;
+
+
+        // connection was upgraded to raw TCP socket
+        NetSocketImpl socket = new NetSocketImpl(conn.vertx, conn.chctx, conn.context, conn.client.getSslHelper(), conn.metrics);
+        socket.metric(conn.metric());
+
+        // Flush out all pending data
+        conn.endReadAndFlush();
+
+        // remove old http handlers and replace the old handler with one that handle plain sockets
+        ChannelPipeline pipeline = conn.chctx.pipeline();
+        ChannelHandler inflater = pipeline.get(HttpContentDecompressor.class);
+        if (inflater != null) {
+          pipeline.remove(inflater);
+        }
+        pipeline.remove("codec");
+        pipeline.replace("handler", "handler",  new VertxNetHandler(socket) {
+          @Override
+          public void channelRead(ChannelHandlerContext chctx, Object msg) throws Exception {
+            if (msg instanceof HttpContent) {
+              if (msg instanceof LastHttpContent) {
+                endResponse((LastHttpContent) msg);
+              }
+              ReferenceCountUtil.release(msg);
+              return;
+            }
+            super.channelRead(chctx, msg);
+          }
+          @Override
+          protected void handleMessage(NetSocketImpl connection, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
+            ByteBuf buf = (ByteBuf) msg;
+            connection.handleMessageReceived(buf);
+          }
+        }.removeHandler(sock -> conn.listener.onDiscard()));
+
+        return socket;
+      }
+    }
+
+    HttpClientResponseImpl beginResponse(HttpResponse resp) {
+      if (conn.metrics != null) {
+        conn.metrics.responseBegin(request.metric(), response);
+      }
+      if (resp.status().code() != 100 && request.method() != io.vertx.core.http.HttpMethod.CONNECT) {
+        // See https://tools.ietf.org/html/rfc7230#section-6.3
+        String responseConnectionHeader = resp.headers().get(HttpHeaders.Names.CONNECTION);
+        io.vertx.core.http.HttpVersion protocolVersion = conn.client.getOptions().getProtocolVersion();
+        String requestConnectionHeader = request.headers().get(HttpHeaders.Names.CONNECTION);
+        // We don't need to protect against concurrent changes on forceClose as it only goes from false -> true
+        if (HttpHeaders.Values.CLOSE.equalsIgnoreCase(responseConnectionHeader) || HttpHeaders.Values.CLOSE.equalsIgnoreCase(requestConnectionHeader)) {
+          // In all cases, if we have a close connection option then we SHOULD NOT treat the connection as persistent
+          close = true;
+        } else if (protocolVersion == io.vertx.core.http.HttpVersion.HTTP_1_0 && !HttpHeaders.Values.KEEP_ALIVE.equalsIgnoreCase(responseConnectionHeader)) {
+          // In the HTTP/1.0 case both request/response need a keep-alive connection header the connection to be persistent
+          // currently Vertx forces the Connection header if keepalive is enabled for 1.0
+          close = true;
+        }
+      }
+      HttpVersion version = HttpUtils.toVertxHttpVersion(resp.protocolVersion());
+      if (version != null) {
+        return response = new HttpClientResponseImpl(request, version, this, resp.status().code(), resp.status().reasonPhrase(), new HeadersAdaptor(resp.headers()));
+      } else {
+        return null;
+      }
+    }
+
+    void endResponse(LastHttpContent trailer) {
+      synchronized (conn) {
+        if (conn.metrics != null) {
+          HttpClientRequestBase req = request;
+          Object reqMetric = req.metric();
+          if (req.exceptionOccurred != null) {
+            conn.metrics.requestReset(reqMetric);
+          } else {
+            conn.metrics.responseEnd(reqMetric, response);
+          }
+        }
+
+        Buffer last = conn.pausedChunk;
+        conn.pausedChunk = null;
+        if (response != null) {
+          response.handleEnd(last, new HeadersAdaptor(trailer.trailingHeaders()));
+        }
+
+        // Also we keep the connection open for an HTTP CONNECT
+        responseEnded = true;
+        if (!conn.client.getOptions().isKeepAlive()) {
+          close = true;
+        }
+        checkLifecycle();
+      }
+    }
+
+    void checkLifecycle() {
+      if (requestEnded && responseEnded) {
+        if (upgraded) {
+          // Do nothing
+        } else if (close) {
+          conn.close();
+        } else {
+          conn.listener.onRecycle(1, true);
+        }
+      }
+    }
+  }
+
+  void handleResponse(HttpResponse resp) {
+    HttpClientResponseImpl response;
+    HttpClientRequestImpl request;
+    synchronized (this) {
+      StreamImpl requestForResponse;
+      if (resp.status().code() == 100) {
+        //If we get a 100 continue it will be followed by the real response later, so we don't remove it yet
+        requestForResponse = inflight.peek();
+      } else {
+        requestForResponse = inflight.poll();
+      }
+      if (requestForResponse == null) {
+        throw new IllegalStateException("No response handler");
+      }
+      currentResponse = requestForResponse;
+      response = currentResponse.beginResponse(resp);
+      request = currentResponse.request;
+    }
+    if (response != null) {
+      request.handleResponse(response);
+    } else {
+      request.handleException(new IllegalStateException("Unsupported HTTP version: " + resp.protocolVersion()));
+    }
+  }
+
+  void handleResponseChunk(Buffer buff) {
+    HttpClientResponseImpl resp;
+    synchronized (this) {
+      if (paused) {
+        if (pausedChunk == null) {
+          pausedChunk = buff.copy();
+        } else {
+          pausedChunk.appendBuffer(buff);
+        }
+        return;
+      } else {
+        if (pausedChunk != null) {
+          buff = pausedChunk.appendBuffer(buff);
+          pausedChunk = null;
+        }
+        resp = currentResponse.response;
+        if (resp == null) {
+          return;
+        }
+      }
+    }
+    resp.handleChunk(buff);
+  }
+
+  void handleResponseEnd(LastHttpContent trailer) {
+    synchronized (this) {
+      StreamImpl resp = currentResponse;
+      currentResponse = null;
+      // We don't signal response end for a 100-continue response as a real response will follow
+      if (resp.response == null || resp.response.statusCode() != 100) {
+        resp.endResponse(trailer);
+      }
+    }
+  }
+
+  public HttpClientMetrics metrics() {
+    return metrics;
+  }
+
+  synchronized void toWebSocket(String requestURI, MultiMap headers, WebsocketVersion vers, String subProtocols,
+                   int maxWebSocketFrameSize, Handler<WebSocket> wsConnect) {
+    if (ws != null) {
+      throw new IllegalStateException("Already websocket");
+    }
+
+    try {
+      URI wsuri = new URI(requestURI);
+      if (!wsuri.isAbsolute()) {
+        // Netty requires an absolute url
+        wsuri = new URI((ssl ? "https:" : "http:") + "//" + host + ":" + port + requestURI);
+      }
+      WebSocketVersion version =
+         WebSocketVersion.valueOf((vers == null ?
+           WebSocketVersion.V13 : vers).toString());
+      HttpHeaders nettyHeaders;
+      if (headers != null) {
+        nettyHeaders = new DefaultHttpHeaders();
+        for (Map.Entry<String, String> entry: headers) {
+          nettyHeaders.add(entry.getKey(), entry.getValue());
+        }
+      } else {
+        nettyHeaders = null;
+      }
+      handshaker = WebSocketClientHandshakerFactory.newHandshaker(wsuri, version, subProtocols, false,
+                                                                  nettyHeaders, maxWebSocketFrameSize,!client.getOptions().isSendUnmaskedFrames(),false);
+      ChannelPipeline p = chctx.pipeline();
+      p.addBefore("handler", "handshakeCompleter", new HandshakeInboundHandler(wsConnect, version != WebSocketVersion.V00));
+      handshaker.handshake(chctx.channel()).addListener(future -> {
+        Handler<Throwable> handler = exceptionHandler();
+        if (!future.isSuccess() && handler != null) {
+          handler.handle(future.cause());
+        }
+      });
+    } catch (Exception e) {
+      handleException(e);
+    }
+  }
+
+  private final class HandshakeInboundHandler extends ChannelInboundHandlerAdapter {
+
+    private final boolean supportsContinuation;
+    private final Handler<WebSocket> wsConnect;
+    private final ContextImpl context;
+    private final Queue<Object> buffered = new ArrayDeque<>();
+    private FullHttpResponse response;
+    private boolean handshaking = true;
+
+    public HandshakeInboundHandler(Handler<WebSocket> wsConnect, boolean supportsContinuation) {
+      this.supportsContinuation = supportsContinuation;
+      this.wsConnect = wsConnect;
+      this.context = vertx.getContext();
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+      super.channelInactive(ctx);
+      // if still handshaking this means we not got any response back from the server and so need to notify the client
+      // about it as otherwise the client would never been notified.
+      if (handshaking) {
+        handleException(new WebSocketHandshakeException("Connection closed while handshake in process"));
+      }
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+      if (handshaker != null && handshaking) {
+        if (msg instanceof HttpResponse) {
+          HttpResponse resp = (HttpResponse) msg;
+          HttpResponseStatus status = resp.status();
+          if (status.code() != 101) {
+            handshaker = null;
+            close();
+            handleException(new WebsocketRejectedException(status.code()));
+            return;
+          }
+          response = new DefaultFullHttpResponse(resp.protocolVersion(), status);
+          response.headers().add(resp.headers());
+        }
+
+        if (msg instanceof HttpContent) {
+          if (response != null) {
+            response.content().writeBytes(((HttpContent) msg).content());
+            if (msg instanceof LastHttpContent) {
+              response.trailingHeaders().add(((LastHttpContent) msg).trailingHeaders());
+              try {
+                handshakeComplete(ctx, response);
+                chctx.pipeline().remove(HandshakeInboundHandler.this);
+                for (; ; ) {
+                  Object m = buffered.poll();
+                  if (m == null) {
+                    break;
+                  }
+                  ctx.fireChannelRead(m);
+                }
+              } catch (WebSocketHandshakeException e) {
+                close();
+                handleException(e);
+              }
+            }
+          }
+        }
+      } else {
+        buffered.add(msg);
+      }
+    }
+
+    private void handleException(Exception e) {
+      handshaking = false;
+      buffered.clear();
+      Handler<Throwable> handler = exceptionHandler();
+      if (handler != null) {
+        context.executeFromIO(() -> {
+          handler.handle(e);
+        });
+      } else {
+        log.error("Error in websocket handshake", e);
+      }
+    }
+
+    private void handshakeComplete(ChannelHandlerContext ctx, FullHttpResponse response) {
+      handshaking = false;
+      ChannelHandler handler = ctx.pipeline().get(HttpContentDecompressor.class);
+      if (handler != null) {
+        // remove decompressor as its not needed anymore once connection was upgraded to websockets
+        ctx.pipeline().remove(handler);
+      }
+      // Need to set context before constructor is called as writehandler registration needs this
+      ContextImpl.setContext(context);
+      WebSocketImpl webSocket = new WebSocketImpl(vertx, Http1xClientConnection.this, supportsContinuation,
+                                                  client.getOptions().getMaxWebsocketFrameSize(),
+                                                  client.getOptions().getMaxWebsocketMessageSize());
+      ws = webSocket;
+      handshaker.finishHandshake(chctx.channel(), response);
+      ws.subProtocol(handshaker.actualSubprotocol());
+      context.executeFromIO(() -> {
+        log.debug("WebSocket handshake complete");
+        if (metrics != null ) {
+          webSocket.setMetric(metrics.connected(endpointMetric, metric(), webSocket));
+        }
+        wsConnect.handle(webSocket);
+      });
+    }
+  }
+
+  @Override
+  public synchronized void handleInterestedOpsChanged() {
+    if (!isNotWritable()) {
+      if (currentRequest != null) {
+        currentRequest.request.handleDrained();
+      } else if (ws != null) {
+        ws.writable();
+      }
+    }
+  }
+
+  public void doPause() {
+    super.doPause();
+    paused = true;
+  }
+
+  public void doResume() {
+    super.doResume();
+    paused = false;
+    if (pausedChunk != null) {
+      context.runOnContext(v -> {
+        if (pausedChunk != null) {
+          Buffer chunk = pausedChunk;
+          pausedChunk = null;
+          currentResponse.response.handleChunk(chunk);
+        }
+      });
+    }
+  }
+
+  synchronized void handleWsFrame(WebSocketFrameInternal frame) {
+    if (ws != null) {
+      ws.handleFrame(frame);
+    }
+  }
+
+  private void retryPending() {
+    StreamImpl stream;
+    while ((stream = pending.poll()) != null) {
+      stream.request.retry();
+    }
+  }
+
+  protected synchronized void handleClosed() {
+    super.handleClosed();
+    if (ws != null) {
+      ws.handleClosed();
+    }
+    retryPending();
+
+    Exception e = new VertxException("Connection was closed");
+
+    // Signal requests failed
+    if (metrics != null) {
+      for (StreamImpl req: inflight) {
+        metrics.requestReset(req.request.metric());
+      }
+      if (currentResponse != null) {
+        metrics.requestReset(currentResponse.request.metric());
+      }
+    }
+
+    // Connection was closed - call exception handlers for any requests in the pipeline or one being currently written
+    for (StreamImpl req: inflight) {
+      if (req != currentRequest) {
+        req.request.handleException(e);
+      }
+    }
+    if (currentRequest != null) {
+      currentRequest.request.handleException(e);
+    } else if (currentResponse != null && currentResponse.response != null) {
+      currentResponse.response.handleException(e);
+    }
+  }
+
+  public ContextImpl getContext() {
+    return super.getContext();
+  }
+
+  @Override
+  protected synchronized void handleException(Throwable e) {
+    super.handleException(e);
+    retryPending();
+    if (currentRequest != null) {
+      currentRequest.request.handleException(e);
+    } else {
+      StreamImpl req = inflight.poll();
+      if (req != null) {
+        req.request.handleException(e);
+      } else if (currentResponse != null && currentResponse.response != null) {
+        currentResponse.response.handleException(e);
+      }
+    }
+  }
+
+  @Override
+  public synchronized void close() {
+    listener.onDiscard();
+    if (handshaker == null) {
+      super.close();
+    } else {
+      // make sure everything is flushed out on close
+      endReadAndFlush();
+      // close the websocket connection by sending a close frame.
+      handshaker.close(chctx.channel(), new CloseWebSocketFrame(1000, null));
+    }
+  }
+
+  @Override
+  public void createStream(HttpClientRequestImpl req, Handler<AsyncResult<HttpClientStream>> handler) {
+    StreamImpl stream = new StreamImpl(this, req, handler);
+    synchronized (this) {
+      if (currentRequest != null) {
+        pending.add(stream);
+        return;
+      }
+      this.currentRequest = stream;
+    }
+    handler.handle(Future.succeededFuture(currentRequest));
+  }
+}
diff --git a/src/main/java/io/vertx/core/http/impl/ClientHandler.java b/src/main/java/io/vertx/core/http/impl/Http1xClientHandler.java
similarity index 71%
rename from src/main/java/io/vertx/core/http/impl/ClientHandler.java
rename to src/main/java/io/vertx/core/http/impl/Http1xClientHandler.java
--- a/src/main/java/io/vertx/core/http/impl/ClientHandler.java
+++ b/src/main/java/io/vertx/core/http/impl/Http1xClientHandler.java
@@ -25,7 +25,8 @@
 import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
+import io.vertx.core.http.HttpVersion;
+import io.vertx.core.http.impl.pool.ConnectionListener;
 import io.vertx.core.http.impl.ws.WebSocketFrameInternal;
 import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
@@ -33,33 +34,43 @@
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-class ClientHandler extends VertxHttpHandler<ClientConnection> {
+class Http1xClientHandler extends VertxHttpHandler<Http1xClientConnection> {
   private boolean closeFrameSent;
   private ContextImpl context;
   private ChannelHandlerContext chctx;
-  private Http1xPool pool;
-  private HttpClientImpl client;
-  private Object endpointMetric;
-  private HttpClientMetrics metrics;
+  private final HttpVersion version;
+  private final String host;
+  private final int port;
+  private final boolean ssl;
+  private final HttpClientImpl client;
+  private final HttpClientMetrics metrics;
+  private final ConnectionListener<HttpClientConnection> listener;
+  private final Object endpointMetric;
 
-  public ClientHandler(ContextImpl context,
-                       Http1xPool pool,
-                       HttpClientImpl client,
-                       Object endpointMetric,
-                       HttpClientMetrics metrics) {
+  public Http1xClientHandler(ConnectionListener<HttpClientConnection> listener,
+                             ContextImpl context,
+                             HttpVersion version,
+                             String host,
+                             int port,
+                             boolean ssl,
+                             HttpClientImpl client,
+                             Object endpointMetric,
+                             HttpClientMetrics metrics) {
     this.context = context;
-    this.pool = pool;
+    this.version = version;
     this.client = client;
+    this.host = host;
+    this.port = port;
+    this.ssl = ssl;
     this.endpointMetric = endpointMetric;
     this.metrics = metrics;
+    this.listener = listener;
   }
 
   @Override
   public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
     chctx = ctx;
-    ClientConnection conn = new ClientConnection(pool.version(), client, endpointMetric, ctx,
-      pool.ssl(), pool.host(), pool.port(), context, pool, metrics);
-    setConnection(conn);
+    Http1xClientConnection conn = new Http1xClientConnection(listener, version, client, endpointMetric, ctx, ssl, host, port, context, metrics);
     if (metrics != null) {
       context.executeFromIO(() -> {
         Object metric = metrics.connected(conn.remoteAddress(), conn.remoteName());
@@ -67,6 +78,7 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
         metrics.endpointConnected(endpointMetric, metric);
       });
     }
+    setConnection(conn);
   }
 
   public ChannelHandlerContext context() {
@@ -74,7 +86,15 @@ public ChannelHandlerContext context() {
   }
 
   @Override
-  protected void handleMessage(ClientConnection conn, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
+  public void channelInactive(ChannelHandlerContext chctx) throws Exception {
+    if (metrics != null) {
+      metrics.endpointDisconnected(endpointMetric, getConnection().metric());
+    }
+    super.channelInactive(chctx);
+  }
+
+  @Override
+  protected void handleMessage(Http1xClientConnection conn, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
     if (msg instanceof HttpObject) {
       HttpObject obj = (HttpObject) msg;
       DecoderResult result = obj.decoderResult();
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xPool.java b/src/main/java/io/vertx/core/http/impl/Http1xPool.java
deleted file mode 100644
--- a/src/main/java/io/vertx/core/http/impl/Http1xPool.java
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
- * Copyright (c) 2011-2013 The original author or authors
- *  ------------------------------------------------------
- *  All rights reserved. This program and the accompanying materials
- *  are made available under the terms of the Eclipse Public License v1.0
- *  and Apache License v2.0 which accompanies this distribution.
- *
- *      The Eclipse Public License is available at
- *      http://www.eclipse.org/legal/epl-v10.html
- *
- *      The Apache License v2.0 is available at
- *      http://www.opensource.org/licenses/apache2.0.php
- *
- *  You may elect to redistribute this code under either of these licenses.
- */
-
-package io.vertx.core.http.impl;
-
-import io.netty.channel.Channel;
-import io.vertx.core.http.HttpClientOptions;
-import io.vertx.core.http.HttpVersion;
-import io.vertx.core.impl.ContextImpl;
-import io.vertx.core.spi.metrics.HttpClientMetrics;
-
-import java.util.ArrayDeque;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Queue;
-import java.util.Set;
-
-/**
- * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
- */
-public class Http1xPool implements ConnectionManager.Pool<ClientConnection> {
-
-  // Pools must locks on the queue object to keep a single lock
-  private final ConnectionManager.ConnQueue queue;
-  private final HttpClientImpl client;
-  private final HttpClientMetrics metrics;
-  private final Map<Channel, HttpClientConnection> connectionMap;
-  private final boolean pipelining;
-  private final boolean keepAlive;
-  private final int pipeliningLimit;
-  private final boolean ssl;
-  private final String host;
-  private final int port;
-  private final HttpVersion version;
-  private final Set<ClientConnection> allConnections = new HashSet<>();
-  private final Queue<ClientConnection> availableConnections = new ArrayDeque<>();
-  private final int maxSockets;
-
-  public Http1xPool(HttpClientImpl client,
-                    HttpClientMetrics metrics,
-                    HttpClientOptions options,
-                    ConnectionManager.ConnQueue queue,
-                    Map<Channel, HttpClientConnection> connectionMap,
-                    HttpVersion version,
-                    int maxSockets,
-                    String host,
-                    int port) {
-    this.queue = queue;
-    this.version = version;
-    this.client = client;
-    this.metrics = metrics;
-    this.pipelining = options.isPipelining();
-    this.keepAlive = options.isKeepAlive();
-    this.pipeliningLimit = options.getPipeliningLimit();
-    this.ssl = options.isSsl();
-    this.connectionMap = connectionMap;
-    this.maxSockets = maxSockets;
-    this.host = host;
-    this.port = port;
-  }
-
-  boolean ssl() {
-    return ssl;
-  }
-
-  String host() {
-    return host;
-  }
-
-  int port() {
-    return port;
-  }
-
-  @Override
-  public HttpVersion version() {
-    // Correct this
-    return version;
-  }
-
-  @Override
-  public ClientConnection pollConnection() {
-    return availableConnections.poll();
-  }
-
-  @Override
-  public boolean canCreateConnection(int connCount) {
-    return connCount < maxSockets;
-  }
-
-  @Override
-  public HttpClientStream createStream(ClientConnection conn) {
-    return conn;
-  }
-
-  public void recycle(ClientConnection conn) {
-    synchronized (queue) {
-      Waiter waiter = queue.getNextWaiter();
-      if (waiter != null) {
-        queue.deliverStream(conn, waiter);
-      } else if (conn.getOutstandingRequestCount() == 0) {
-        // Return to set of available from here to not return it several times
-        availableConnections.add(conn);
-      }
-    }
-  }
-
-  void requestEnded(ClientConnection conn) {
-    ContextImpl context = conn.getContext();
-    context.runOnContext(v -> {
-      if (pipelining && conn.getOutstandingRequestCount() < pipeliningLimit) {
-        recycle(conn);
-      }
-    });
-  }
-
-  void responseEnded(ClientConnection conn, boolean close) {
-    if (!keepAlive || close) {
-      conn.close();
-    } else {
-      ContextImpl ctx = conn.getContext();
-      ctx.runOnContext(v -> {
-        if (conn.getCurrentRequest() == null) {
-          recycle(conn);
-        }
-      });
-    }
-  }
-
-  void createConn(ContextImpl context, Channel ch, Waiter waiter) {
-    ClientHandler handler = new ClientHandler(
-      context,
-      this,
-      client,
-      queue.metric,
-      metrics);
-    handler.addHandler(conn -> {
-      synchronized (queue) {
-        allConnections.add(conn);
-      }
-      connectionMap.put(ch, conn);
-    });
-    handler.removeHandler(this::connectionClosed);
-    ch.pipeline().addLast("handler", handler);
-    ClientConnection conn = handler.getConnection();
-    context.executeFromIO(() -> {
-      waiter.handleConnection(conn);
-      queue.deliverStream(conn, waiter);
-    });
-  }
-
-  // Called if the connection is actually closed, OR the connection attempt failed - in the latter case
-  // conn will be null
-  // The connection has been closed - tell the pool about it, this allows the pool to create more
-  // connections. Note the pool doesn't actually remove the connection, when the next person to get a connection
-  // gets the closed on, they will check if it's closed and if so get another one.
-  private synchronized void connectionClosed(ClientConnection conn) {
-    synchronized (queue) {
-      connectionMap.remove(conn.channel());
-      allConnections.remove(conn);
-      availableConnections.remove(conn);
-      queue.connectionClosed();
-    }
-    if (metrics != null) {
-      metrics.endpointDisconnected(queue.metric, conn.metric());
-    }
-  }
-
-  public void closeAllConnections() {
-    Set<ClientConnection> copy;
-    synchronized (this) {
-      copy = new HashSet<>(allConnections);
-      allConnections.clear();
-    }
-    // Close outside sync block to avoid deadlock
-    for (ClientConnection conn : copy) {
-      try {
-        conn.close();
-      } catch (Throwable t) {
-        ConnectionManager.log.error("Failed to close connection", t);
-      }
-    }
-  }
-
-  void removeChannel(Channel channel) {
-    connectionMap.remove(channel);
-  }
-}
diff --git a/src/main/java/io/vertx/core/http/impl/ServerConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java
similarity index 95%
rename from src/main/java/io/vertx/core/http/impl/ServerConnection.java
rename to src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java
--- a/src/main/java/io/vertx/core/http/impl/ServerConnection.java
+++ b/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java
@@ -81,9 +81,9 @@
  *
  * @author <a href="http://tfox.org">Tim Fox</a>
  */
-public class ServerConnection extends Http1xConnectionBase implements HttpConnection {
+public class Http1xServerConnection extends Http1xConnectionBase implements HttpConnection {
 
-  private static final Logger log = LoggerFactory.getLogger(ServerConnection.class);
+  private static final Logger log = LoggerFactory.getLogger(Http1xServerConnection.class);
 
   private static final Handler<HttpServerRequest> NULL_REQUEST_HANDLER = req -> {};
 
@@ -111,13 +111,13 @@ public class ServerConnection extends Http1xConnectionBase implements HttpConnec
   // queuing == true <=> (paused || (pendingResponse != null && msg instanceof HttpRequest) || !pending.isEmpty())
   private boolean queueing;
 
-  public ServerConnection(VertxInternal vertx,
-                   SSLHelper sslHelper,
-                   HttpServerOptions options,
-                   ChannelHandlerContext channel,
-                   ContextImpl context,
-                   String serverOrigin,
-                   HttpServerMetrics metrics) {
+  public Http1xServerConnection(VertxInternal vertx,
+                                SSLHelper sslHelper,
+                                HttpServerOptions options,
+                                ChannelHandlerContext channel,
+                                ContextImpl context,
+                                String serverOrigin,
+                                HttpServerMetrics metrics) {
     super(vertx, channel, context);
     this.serverOrigin = serverOrigin;
     this.options = options;
@@ -221,7 +221,7 @@ ServerWebSocket upgrade(HttpServerRequest request, HttpRequest nettyReq) {
     if (ws != null) {
       return ws;
     }
-    ServerHandler serverHandler = (ServerHandler) chctx.pipeline().get("handler");
+    Http1xServerHandler serverHandler = (Http1xServerHandler) chctx.pipeline().get("handler");
     handshaker = serverHandler.createHandshaker(this, chctx.channel(), nettyReq);
     if (handshaker == null) {
       throw new IllegalStateException("Can't upgrade this request");
@@ -507,7 +507,7 @@ private void checkNextTick() {
           }
           if (channelPaused && pending.isEmpty()) {
             //Resume the actual channel
-            ServerConnection.super.doResume();
+            Http1xServerConnection.super.doResume();
             channelPaused = false;
           }
         }
diff --git a/src/main/java/io/vertx/core/http/impl/ServerHandler.java b/src/main/java/io/vertx/core/http/impl/Http1xServerHandler.java
similarity index 83%
rename from src/main/java/io/vertx/core/http/impl/ServerHandler.java
rename to src/main/java/io/vertx/core/http/impl/Http1xServerHandler.java
--- a/src/main/java/io/vertx/core/http/impl/ServerHandler.java
+++ b/src/main/java/io/vertx/core/http/impl/Http1xServerHandler.java
@@ -23,9 +23,9 @@
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-public class ServerHandler extends VertxHttpHandler<ServerConnection> {
+public class Http1xServerHandler extends VertxHttpHandler<Http1xServerConnection> {
 
-  private static final Logger log = LoggerFactory.getLogger(ServerHandler.class);
+  private static final Logger log = LoggerFactory.getLogger(Http1xServerHandler.class);
 
   private final SSLHelper sslHelper;
   private final HttpServerOptions options;
@@ -33,7 +33,7 @@ public class ServerHandler extends VertxHttpHandler<ServerConnection> {
   private final HttpServerMetrics metrics;
   private final HandlerHolder<HttpHandlers> holder;
 
-  public ServerHandler(SSLHelper sslHelper, HttpServerOptions options, String serverOrigin, HandlerHolder<HttpHandlers> holder, HttpServerMetrics metrics) {
+  public Http1xServerHandler(SSLHelper sslHelper, HttpServerOptions options, String serverOrigin, HandlerHolder<HttpHandlers> holder, HttpServerMetrics metrics) {
     this.holder = holder;
     this.metrics = metrics;
     this.sslHelper = sslHelper;
@@ -44,7 +44,7 @@ public ServerHandler(SSLHelper sslHelper, HttpServerOptions options, String serv
   @Override
   public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
     super.handlerAdded(ctx);
-    ServerConnection conn = new ServerConnection(holder.context.owner(),
+    Http1xServerConnection conn = new Http1xServerConnection(holder.context.owner(),
       sslHelper,
       options,
       ctx,
@@ -65,11 +65,11 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
   }
 
   @Override
-  protected void handleMessage(ServerConnection conn, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
+  protected void handleMessage(Http1xServerConnection conn, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
     conn.handleMessage(msg);
   }
 
-  WebSocketServerHandshaker createHandshaker(ServerConnection conn, Channel ch, HttpRequest request) {
+  WebSocketServerHandshaker createHandshaker(Http1xServerConnection conn, Channel ch, HttpRequest request) {
     // As a fun part, Firefox 6.0.2 supports Websockets protocol '7'. But,
     // it doesn't send a normal 'Connection: Upgrade' header. Instead it
     // sends: 'Connection: keep-alive, Upgrade'. Brilliant.
diff --git a/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java
--- a/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java
+++ b/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java
@@ -18,6 +18,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpResponseStatus;
@@ -27,16 +28,10 @@
 import io.netty.handler.codec.http2.Http2Exception;
 import io.netty.handler.codec.http2.Http2Headers;
 import io.netty.handler.codec.http2.Http2Stream;
-import io.vertx.core.Context;
-import io.vertx.core.Handler;
-import io.vertx.core.MultiMap;
-import io.vertx.core.VertxException;
+import io.vertx.core.*;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.CaseInsensitiveHeaders;
-import io.vertx.core.http.HttpClientRequest;
-import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.HttpVersion;
-import io.vertx.core.http.StreamResetException;
+import io.vertx.core.http.*;
+import io.vertx.core.http.impl.pool.ConnectionListener;
 import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.net.NetSocket;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
@@ -50,63 +45,76 @@
  */
 class Http2ClientConnection extends Http2ConnectionBase implements HttpClientConnection {
 
-  final Http2Pool http2Pool;
+  private final ConnectionListener<HttpClientConnection> listener;
+  private final HttpClientImpl client;
   final HttpClientMetrics metrics;
   final Object queueMetric;
-  int streamCount;
 
-  public Http2ClientConnection(Http2Pool http2Pool,
+  public Http2ClientConnection(ConnectionListener<HttpClientConnection> listener,
                                Object queueMetric,
+                               HttpClientImpl client,
                                ContextImpl context,
                                VertxHttp2ConnectionHandler connHandler,
                                HttpClientMetrics metrics) {
     super(context, connHandler);
-    this.http2Pool = http2Pool;
     this.metrics = metrics;
     this.queueMetric = queueMetric;
+    this.client = client;
+    this.listener = listener;
   }
 
   @Override
-  public HttpClientMetrics metrics() {
-    return metrics;
+  synchronized void onGoAwaySent(int lastStreamId, long errorCode, ByteBuf debugData) {
+    listener.onDiscard();
+    super.onGoAwaySent(lastStreamId, errorCode, debugData);
   }
 
   @Override
-  void onGoAwaySent(int lastStreamId, long errorCode, ByteBuf debugData) {
-    http2Pool.discard(Http2ClientConnection.this);
+  synchronized void onGoAwayReceived(int lastStreamId, long errorCode, ByteBuf debugData) {
+    listener.onDiscard();
+    super.onGoAwayReceived(lastStreamId, errorCode, debugData);
   }
 
   @Override
-  void onGoAwayReceived(int lastStreamId, long errorCode, ByteBuf debugData) {
-    super.onGoAwayReceived(lastStreamId, errorCode, debugData);
-    http2Pool.discard(Http2ClientConnection.this);
+  public Channel channel() {
+    return chctx.channel();
   }
 
   @Override
-  void onStreamClosed(Http2Stream nettyStream) {
-    super.onStreamClosed(nettyStream);
-    http2Pool.recycle(Http2ClientConnection.this);
+  protected void concurrencyChanged(long concurrency) {
+    int limit = client.getOptions().getHttp2MultiplexingLimit();
+    if (limit > 0) {
+      concurrency = Math.min(concurrency, limit);
+    }
+    listener.onConcurrencyChange(concurrency);
   }
 
-  synchronized HttpClientStream createStream() throws Http2Exception {
-    Http2Connection conn = handler.connection();
-    Http2Stream stream = conn.local().createStream(conn.local().incrementAndGetNextStreamId(), false);
-    boolean writable = handler.encoder().flowController().isWritable(stream);
-    Http2ClientStream clientStream = new Http2ClientStream(this, stream, writable);
-    streams.put(clientStream.stream.id(), clientStream);
-    return clientStream;
+  @Override
+  public HttpClientMetrics metrics() {
+    return metrics;
   }
 
   @Override
-  public synchronized void handleClosed() {
-    http2Pool.discard(this);
-    super.handleClosed();
+  void onStreamClosed(Http2Stream nettyStream) {
+    super.onStreamClosed(nettyStream);
   }
 
   @Override
-  public boolean isValid() {
-    Http2Connection conn = handler.connection();
-    return !isClosed() && !conn.goAwaySent() && !conn.goAwayReceived();
+  public void createStream(HttpClientRequestImpl req, Handler<AsyncResult<HttpClientStream>> completionHandler) {
+    Future<HttpClientStream> fut;
+    synchronized (this) {
+      try {
+        Http2Connection conn = handler.connection();
+        Http2Stream stream = conn.local().createStream(conn.local().incrementAndGetNextStreamId(), false);
+        boolean writable = handler.encoder().flowController().isWritable(stream);
+        Http2ClientStream clientStream = new Http2ClientStream(this, req, stream, writable);
+        streams.put(clientStream.stream.id(), clientStream);
+        fut = Future.succeededFuture(clientStream);
+      } catch (Http2Exception e) {
+        fut = Future.failedFuture(e);
+      }
+    }
+    completionHandler.handle(fut);
   }
 
   @Override
@@ -133,7 +141,7 @@ public synchronized void onPushPromiseRead(ChannelHandlerContext ctx, int stream
           MultiMap headersMap = new Http2HeadersAdaptor(headers);
           Http2Stream promisedStream = handler.connection().stream(promisedStreamId);
           int port = remoteAddress().port();
-          HttpClientRequestPushPromise pushReq = new HttpClientRequestPushPromise(this, promisedStream, http2Pool.client, isSsl(), method, rawMethod, uri, host, port, headersMap);
+          HttpClientRequestPushPromise pushReq = new HttpClientRequestPushPromise(this, promisedStream, client, isSsl(), method, rawMethod, uri, host, port, headersMap);
           if (metrics != null) {
             pushReq.metric(metrics.responsePushed(queueMetric, metric(), localAddress(), remoteAddress(), pushReq));
           }
@@ -148,15 +156,11 @@ public synchronized void onPushPromiseRead(ChannelHandlerContext ctx, int stream
 
   static class Http2ClientStream extends VertxHttp2Stream<Http2ClientConnection> implements HttpClientStream {
 
-    private HttpClientRequestBase request;
+    private final HttpClientRequestBase request;
     private HttpClientResponseImpl response;
     private boolean requestEnded;
     private boolean responseEnded;
 
-    public Http2ClientStream(Http2ClientConnection conn, Http2Stream stream, boolean writable) throws Http2Exception {
-      this(conn, null, stream, writable);
-    }
-
     public Http2ClientStream(Http2ClientConnection conn, HttpClientRequestBase request, Http2Stream stream, boolean writable) throws Http2Exception {
       super(conn, stream, writable);
       this.request = request;
@@ -208,6 +212,13 @@ void handleReset(long errorCode) {
 
     @Override
     void handleClose() {
+      // commented to be used later when we properly define the HTTP/2 connection expiration from the pool
+      // boolean disposable = conn.streams.isEmpty();
+      if (request instanceof HttpClientRequestImpl) {
+        conn.listener.onRecycle(1, false);
+      } /* else {
+        conn.listener.onRecycle(0, dispable);
+      } */
       if (!responseEnded) {
         responseEnded = true;
         if (conn.metrics != null) {
@@ -313,7 +324,7 @@ public void writeHeadWithContent(HttpMethod method, String rawMethod, String uri
           h.add(Http2HeadersAdaptor.toLowerCase(header.getKey()), header.getValue());
         }
       }
-      if (conn.http2Pool.client.getOptions().isTryUseCompression() && h.get(HttpHeaderNames.ACCEPT_ENCODING) == null) {
+      if (conn.client.getOptions().isTryUseCompression() && h.get(HttpHeaderNames.ACCEPT_ENCODING) == null) {
         h.set(HttpHeaderNames.ACCEPT_ENCODING, DEFLATE_GZIP);
       }
       if (conn.metrics != null) {
@@ -360,8 +371,7 @@ public boolean isNotWritable() {
     }
 
     @Override
-    public void beginRequest(HttpClientRequestImpl request) {
-      this.request = request;
+    public void beginRequest() {
     }
 
     @Override
diff --git a/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java b/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java
--- a/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java
+++ b/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java
@@ -20,7 +20,6 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
-import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http2.Http2Connection;
 import io.netty.handler.codec.http2.Http2Exception;
@@ -81,21 +80,25 @@ static ByteBuf safeBuffer(ByteBuf buf, ByteBufAllocator allocator) {
   protected final ChannelHandlerContext handlerContext;
   protected final VertxHttp2ConnectionHandler handler;
   private boolean shutdown;
-  private Handler<io.vertx.core.http.Http2Settings> clientSettingsHandler;
+  private Handler<io.vertx.core.http.Http2Settings> remoteSettingsHandler;
   private final ArrayDeque<Runnable> updateSettingsHandlers = new ArrayDeque<>(4);
   private final ArrayDeque<Handler<AsyncResult<Buffer>>> pongHandlers = new ArrayDeque<>();
-  private Http2Settings serverSettings = new Http2Settings();
+  private Http2Settings localSettings = new Http2Settings();
+  private Http2Settings remoteSettings;
   private Handler<GoAway> goAwayHandler;
   private Handler<Void> shutdownHandler;
   private Handler<Buffer> pingHandler;
   private boolean closed;
+  private boolean goneAway;
   private int windowSize;
+  private long maxConcurrentStreams;
 
   public Http2ConnectionBase(ContextImpl context, VertxHttp2ConnectionHandler handler) {
     super(context.owner(), handler.context(), context);
     this.handler = handler;
     this.handlerContext = chctx;
     this.windowSize = handler.connection().local().flowController().windowSize(handler.connection().connectionStream());
+    this.maxConcurrentStreams = io.vertx.core.http.Http2Settings.DEFAULT_MAX_CONCURRENT_STREAMS;
   }
 
   VertxInternal vertx() {
@@ -165,18 +168,25 @@ synchronized void onStreamClosed(Http2Stream stream) {
     }
   }
 
-  void onGoAwaySent(int lastStreamId, long errorCode, ByteBuf debugData) {
+  synchronized void onGoAwaySent(int lastStreamId, long errorCode, ByteBuf debugData) {
+    if (!goneAway) {
+      goneAway = true;
+      checkShutdownHandler();
+    }
   }
 
   synchronized void onGoAwayReceived(int lastStreamId, long errorCode, ByteBuf debugData) {
-    Handler<GoAway> handler = goAwayHandler;
-    if (handler != null) {
-      Buffer buffer = Buffer.buffer(debugData);
-      context.executeFromIO(() -> {
-        handler.handle(new GoAway().setErrorCode(errorCode).setLastStreamId(lastStreamId).setDebugData(buffer));
-      });
+    if (!goneAway) {
+      goneAway = true;
+      Handler<GoAway> handler = goAwayHandler;
+      if (handler != null) {
+        Buffer buffer = Buffer.buffer(debugData);
+        context.executeFromIO(() -> {
+          handler.handle(new GoAway().setErrorCode(errorCode).setLastStreamId(lastStreamId).setDebugData(buffer));
+        });
+      }
+      checkShutdownHandler();
     }
-    checkShutdownHandler();
   }
 
   // Http2FrameListener
@@ -200,13 +210,37 @@ public synchronized void onSettingsAckRead(ChannelHandlerContext ctx) {
     }
   }
 
+  protected void onConnect() {
+  }
+
+  protected void concurrencyChanged(long concurrency) {
+  }
+
   @Override
-  public synchronized void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) {
-    Handler<io.vertx.core.http.Http2Settings> handler = clientSettingsHandler;
-    if (handler != null) {
-      context.executeFromIO(() -> {
-        handler.handle(HttpUtils.toVertxSettings(settings));
-      });
+  public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) {
+    boolean changed;
+    Long val = settings.maxConcurrentStreams();
+    if (val != null) {
+      if (remoteSettings != null) {
+        changed = val != maxConcurrentStreams;
+      } else {
+        changed = false;
+      }
+      maxConcurrentStreams = val;
+    } else {
+      changed = false;
+    }
+    remoteSettings = settings;
+    synchronized (this) {
+      Handler<io.vertx.core.http.Http2Settings> handler = remoteSettingsHandler;
+      if (handler != null) {
+        context.executeFromIO(() -> {
+          handler.handle(HttpUtils.toVertxSettings(settings));
+        });
+      }
+    }
+    if (changed) {
+      concurrencyChanged(maxConcurrentStreams);
     }
   }
 
@@ -311,7 +345,7 @@ public synchronized HttpConnection goAway(long errorCode, int lastStreamId, Buff
       throw new IllegalArgumentException();
     }
     if (lastStreamId < 0) {
-      throw new IllegalArgumentException();
+      lastStreamId = handler.connection().remote().lastStreamCreated();
     }
     handler.writeGoAway(errorCode, lastStreamId, debugData != null ? debugData.getByteBuf() : Unpooled.EMPTY_BUFFER);
     return this;
@@ -357,12 +391,13 @@ public void close() {
 
   @Override
   public synchronized HttpConnection remoteSettingsHandler(Handler<io.vertx.core.http.Http2Settings> handler) {
-    clientSettingsHandler = handler;
+    remoteSettingsHandler = handler;
     return this;
   }
 
   @Override
   public synchronized io.vertx.core.http.Http2Settings remoteSettings() {
+    /*
     io.vertx.core.http.Http2Settings a = new io.vertx.core.http.Http2Settings();
     a.setPushEnabled(handler.connection().remote().allowPushTo());
     a.setMaxConcurrentStreams((long) handler.connection().local().maxActiveStreams());
@@ -371,11 +406,13 @@ public synchronized io.vertx.core.http.Http2Settings remoteSettings() {
     a.setMaxFrameSize(handler.encoder().configuration().frameSizePolicy().maxFrameSize());
     a.setInitialWindowSize(handler.encoder().flowController().initialWindowSize());
     return a;
+    */
+    return HttpUtils.toVertxSettings(remoteSettings);
   }
 
   @Override
   public synchronized io.vertx.core.http.Http2Settings settings() {
-    return HttpUtils.toVertxSettings(serverSettings);
+    return HttpUtils.toVertxSettings(localSettings);
   }
 
   @Override
@@ -403,7 +440,7 @@ protected synchronized void updateSettings(Http2Settings settingsUpdate, Handler
       if (fut.isSuccess()) {
         synchronized (Http2ConnectionBase.this) {
           updateSettingsHandlers.add(() -> {
-            serverSettings.putAll(settingsUpdate);
+            localSettings.putAll(settingsUpdate);
             if (completionHandler != null) {
               completionContext.runOnContext(v -> {
                 completionHandler.handle(Future.succeededFuture());
diff --git a/src/main/java/io/vertx/core/http/impl/Http2Pool.java b/src/main/java/io/vertx/core/http/impl/Http2Pool.java
deleted file mode 100644
--- a/src/main/java/io/vertx/core/http/impl/Http2Pool.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/*
- * Copyright (c) 2011-2013 The original author or authors
- *  ------------------------------------------------------
- *  All rights reserved. This program and the accompanying materials
- *  are made available under the terms of the Eclipse Public License v1.0
- *  and Apache License v2.0 which accompanies this distribution.
- *
- *      The Eclipse Public License is available at
- *      http://www.eclipse.org/legal/epl-v10.html
- *
- *      The Apache License v2.0 is available at
- *      http://www.opensource.org/licenses/apache2.0.php
- *
- *  You may elect to redistribute this code under either of these licenses.
- */
-
-package io.vertx.core.http.impl;
-
-import io.netty.channel.Channel;
-import io.netty.handler.codec.http2.Http2Exception;
-import io.vertx.core.http.HttpVersion;
-import io.vertx.core.impl.ContextImpl;
-import io.vertx.core.spi.metrics.HttpClientMetrics;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
- */
-class Http2Pool implements ConnectionManager.Pool<Http2ClientConnection> {
-
-  // Pools must locks on the queue object to keep a single lock
-  private final ConnectionManager.ConnQueue queue;
-  private final Set<Http2ClientConnection> allConnections = new HashSet<>();
-  private final Map<Channel, ? super Http2ClientConnection> connectionMap;
-  final HttpClientImpl client;
-  final HttpClientMetrics metrics;
-  final int maxConcurrency;
-  final boolean logEnabled;
-  final int maxSockets;
-  final int windowSize;
-
-  public Http2Pool(ConnectionManager.ConnQueue queue, HttpClientImpl client, HttpClientMetrics metrics,
-                   Map<Channel, ? super Http2ClientConnection> connectionMap,
-                   int maxConcurrency, boolean logEnabled, int maxSize, int windowSize) {
-    this.queue = queue;
-    this.client = client;
-    this.metrics = metrics;
-    this.connectionMap = connectionMap;
-    this.maxConcurrency = maxConcurrency;
-    this.logEnabled = logEnabled;
-    this.maxSockets = maxSize;
-    this.windowSize = windowSize;
-  }
-
-  @Override
-  public HttpVersion version() {
-    return HttpVersion.HTTP_2;
-  }
-
-  @Override
-  public boolean canCreateConnection(int connCount) {
-    // We create at most one connection concurrently when all others when
-    // all others are busy
-    return connCount == allConnections.size() && connCount < maxSockets;
-  }
-
-  @Override
-  public Http2ClientConnection pollConnection() {
-    for (Http2ClientConnection conn : allConnections) {
-      if (canReserveStream(conn)) {
-        conn.streamCount++;
-        return conn;
-      }
-    }
-    return null;
-  }
-
-  void createConn(ContextImpl context, Channel ch, Waiter waiter, boolean upgrade) throws Http2Exception {
-    synchronized (queue) {
-      VertxHttp2ConnectionHandler<Http2ClientConnection> handler = new VertxHttp2ConnectionHandlerBuilder<Http2ClientConnection>(ch)
-          .connectionMap(connectionMap)
-          .server(false)
-          .clientUpgrade(upgrade)
-          .useCompression(client.getOptions().isTryUseCompression())
-          .initialSettings(client.getOptions().getInitialSettings())
-          .connectionFactory(connHandler -> {
-            Http2ClientConnection conn = new Http2ClientConnection(Http2Pool.this, queue.metric, context, connHandler, metrics);
-            if (metrics != null) {
-              Object metric = metrics.connected(conn.remoteAddress(), conn.remoteName());
-              conn.metric(metric);
-            }
-            return conn;
-          })
-          .logEnabled(logEnabled)
-          .build();
-      Http2ClientConnection conn = handler.connection;
-      if (metrics != null) {
-        metrics.endpointConnected(queue.metric, conn.metric());
-      }
-      allConnections.add(conn);
-      if (windowSize > 0) {
-        conn.setWindowSize(windowSize);
-      }
-      conn.streamCount++;
-      waiter.handleConnection(conn); // Should make same tests than in deliverRequest
-      queue.deliverStream(conn, waiter);
-      checkPending(conn);
-    }
-  }
-
-  private boolean canReserveStream(Http2ClientConnection handler) {
-    int maxConcurrentStreams = Math.min(handler.handler.connection().local().maxActiveStreams(), maxConcurrency);
-    return handler.streamCount < maxConcurrentStreams;
-  }
-
-  void checkPending(Http2ClientConnection conn) {
-    synchronized (queue) {
-      Waiter waiter;
-      while (canReserveStream(conn) && (waiter = queue.getNextWaiter()) != null) {
-        conn.streamCount++;
-        queue.deliverStream(conn, waiter);
-      }
-    }
-  }
-
-  void discard(Http2ClientConnection conn) {
-    synchronized (queue) {
-      if (allConnections.remove(conn)) {
-        queue.connectionClosed();
-      }
-    }
-    if (metrics != null) {
-      metrics.endpointDisconnected(queue.metric, conn.metric());
-    }
-  }
-
-  @Override
-  public void recycle(Http2ClientConnection conn) {
-    synchronized (queue) {
-      conn.streamCount--;
-      checkPending(conn);
-    }
-  }
-
-  @Override
-  public HttpClientStream createStream(Http2ClientConnection conn) throws Exception {
-    return conn.createStream();
-  }
-
-  @Override
-  public void closeAllConnections() {
-    List<Http2ClientConnection> toClose;
-    synchronized (queue) {
-      toClose = new ArrayList<>(allConnections);
-    }
-    // Close outside sync block to avoid deadlock
-    toClose.forEach(Http2ConnectionBase::close);
-  }
-}
diff --git a/src/main/java/io/vertx/core/http/impl/Http2ServerConnection.java b/src/main/java/io/vertx/core/http/impl/Http2ServerConnection.java
--- a/src/main/java/io/vertx/core/http/impl/Http2ServerConnection.java
+++ b/src/main/java/io/vertx/core/http/impl/Http2ServerConnection.java
@@ -30,11 +30,7 @@
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.HttpServerOptions;
-import io.vertx.core.http.HttpServerRequest;
-import io.vertx.core.http.HttpServerResponse;
-import io.vertx.core.http.StreamResetException;
+import io.vertx.core.http.*;
 import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.spi.metrics.HttpServerMetrics;
 
diff --git a/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java b/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java
@@ -0,0 +1,346 @@
+/*
+ * Copyright (c) 2011-2013 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ *     The Eclipse Public License is available at
+ *     http://www.eclipse.org/legal/epl-v10.html
+ *
+ *     The Apache License v2.0 is available at
+ *     http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.vertx.core.http.impl;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.http.*;
+import io.netty.handler.logging.LoggingHandler;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Handler;
+import io.vertx.core.http.HttpClientOptions;
+import io.vertx.core.http.HttpVersion;
+import io.vertx.core.http.impl.pool.ConnectionListener;
+import io.vertx.core.http.impl.pool.ConnectionProvider;
+import io.vertx.core.impl.ContextImpl;
+import io.vertx.core.net.ProxyType;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.core.net.impl.ChannelProvider;
+import io.vertx.core.net.impl.ProxyChannelProvider;
+import io.vertx.core.net.impl.SSLHelper;
+import io.vertx.core.spi.metrics.HttpClientMetrics;
+
+import javax.net.ssl.SSLHandshakeException;
+
+/**
+ * Performs the channel configuration and connection according to the client options and the protocol version.
+ *
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ */
+class HttpChannelConnector implements ConnectionProvider<HttpClientConnection> {
+
+  private final HttpClientImpl client;
+  private final HttpClientOptions options;
+  private final HttpClientMetrics metrics;
+  private final SSLHelper sslHelper;
+  private final HttpVersion version;
+  private final long weight;
+  private final long http1Weight;
+  private final long http2Weight;
+  private final long http2MaxConcurrency;
+  private final long http1MaxConcurrency;
+  private final boolean ssl;
+  private final String peerHost;
+  private final String host;
+  private final int port;
+  private final Object metric;
+
+  HttpChannelConnector(HttpClientImpl client,
+                       Object metric,
+                       HttpVersion version,
+                       boolean ssl,
+                       String peerHost,
+                       String host,
+                       int port) {
+    this.client = client;
+    this.metric = metric;
+    this.options = client.getOptions();
+    this.metrics = client.metrics();
+    this.sslHelper = client.getSslHelper();
+    this.version = version;
+    this.http1Weight = client.getOptions().getHttp2MaxPoolSize();
+    this.http2Weight = client.getOptions().getMaxPoolSize();
+    this.weight = version == HttpVersion.HTTP_2 ? http2Weight : http1Weight;
+    this.http2MaxConcurrency = client.getOptions().getHttp2MultiplexingLimit() <= 0 ? Long.MAX_VALUE : client.getOptions().getHttp2MultiplexingLimit();
+    this.http1MaxConcurrency = client.getOptions().isPipelining() ? client.getOptions().getPipeliningLimit() : 1;
+    this.ssl = ssl;
+    this.peerHost = peerHost;
+    this.host = host;
+    this.port = port;
+  }
+
+  @Override
+  public void close(HttpClientConnection conn) {
+    conn.close();
+  }
+
+  public long connect(
+    ConnectionListener<HttpClientConnection> listener,
+    ContextImpl context) {
+
+    Bootstrap bootstrap = new Bootstrap();
+    bootstrap.group(context.nettyEventLoop());
+    bootstrap.channel(client.getVertx().transport().channelType(false));
+
+    applyConnectionOptions(bootstrap);
+
+    ChannelProvider channelProvider;
+    // http proxy requests are handled in HttpClientImpl, everything else can use netty proxy handler
+    if (options.getProxyOptions() == null || !ssl && options.getProxyOptions().getType()== ProxyType.HTTP ) {
+      channelProvider = ChannelProvider.INSTANCE;
+    } else {
+      channelProvider = ProxyChannelProvider.INSTANCE;
+    }
+
+    boolean useAlpn = options.isUseAlpn();
+    Handler<Channel> channelInitializer = ch -> {
+
+      // Configure pipeline
+      ChannelPipeline pipeline = ch.pipeline();
+      if (ssl) {
+        SslHandler sslHandler = new SslHandler(sslHelper.createEngine(client.getVertx(), peerHost, port, options.isForceSni() ? peerHost : null));
+        ch.pipeline().addLast("ssl", sslHandler);
+        // TCP connected, so now we must do the SSL handshake
+        sslHandler.handshakeFuture().addListener(fut -> {
+          if (fut.isSuccess()) {
+            String protocol = sslHandler.applicationProtocol();
+            if (useAlpn) {
+              if ("h2".equals(protocol)) {
+                applyHttp2ConnectionOptions(ch.pipeline());
+                http2Connected(listener, context, ch);
+              } else {
+                applyHttp1xConnectionOptions(ch.pipeline());
+                HttpVersion fallbackProtocol = "http/1.0".equals(protocol) ?
+                  HttpVersion.HTTP_1_0 : HttpVersion.HTTP_1_1;
+                http1xConnected(listener, fallbackProtocol, host, port, true, context, ch);
+              }
+            } else {
+              applyHttp1xConnectionOptions(ch.pipeline());
+              http1xConnected(listener, version, host, port, true, context, ch);
+            }
+          } else {
+            handshakeFailure(context, ch, fut.cause(), listener);
+          }
+        });
+      } else {
+        if (version == HttpVersion.HTTP_2) {
+          if (client.getOptions().isHttp2ClearTextUpgrade()) {
+            HttpClientCodec httpCodec = new HttpClientCodec();
+            class UpgradeRequestHandler extends ChannelInboundHandlerAdapter {
+              @Override
+              public void channelActive(ChannelHandlerContext ctx) throws Exception {
+                DefaultFullHttpRequest upgradeRequest =
+                    new DefaultFullHttpRequest(io.netty.handler.codec.http.HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
+                String hostHeader = peerHost;
+                if (port != 80) {
+                  hostHeader += ":" + port;
+                }
+                upgradeRequest.headers().set(HttpHeaderNames.HOST, hostHeader);
+                ctx.writeAndFlush(upgradeRequest);
+                ctx.fireChannelActive();
+              }
+              @Override
+              public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+                if (msg instanceof LastHttpContent) {
+                  ChannelPipeline p = ctx.pipeline();
+                  p.remove(httpCodec);
+                  p.remove(this);
+                  // Upgrade handler will remove itself
+                  applyHttp1xConnectionOptions(ch.pipeline());
+                  http1xConnected(listener, HttpVersion.HTTP_1_1, host, port, false, context, ch);
+                }
+              }
+              @Override
+              public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+                super.userEventTriggered(ctx, evt);
+                if (evt == HttpClientUpgradeHandler.UpgradeEvent.UPGRADE_SUCCESSFUL) {
+                  ctx.pipeline().remove(this);
+                  // Upgrade handler will remove itself
+                }
+              }
+            }
+            VertxHttp2ClientUpgradeCodec upgradeCodec = new VertxHttp2ClientUpgradeCodec(client.getOptions().getInitialSettings()) {
+              @Override
+              public void upgradeTo(ChannelHandlerContext ctx, FullHttpResponse upgradeResponse) throws Exception {
+                applyHttp2ConnectionOptions(pipeline);
+                http2Connected(listener, context, ch);
+              }
+            };
+            HttpClientUpgradeHandler upgradeHandler = new HttpClientUpgradeHandler(httpCodec, upgradeCodec, 65536);
+            ch.pipeline().addLast(httpCodec, upgradeHandler, new UpgradeRequestHandler());
+          } else {
+            applyHttp2ConnectionOptions(pipeline);
+          }
+        } else {
+          applyHttp1xConnectionOptions(pipeline);
+        }
+      }
+    };
+
+    Handler<AsyncResult<Channel>> channelHandler = res -> {
+
+      if (res.succeeded()) {
+        Channel ch = res.result();
+        if (!ssl) {
+          if (ch.pipeline().get(HttpClientUpgradeHandler.class) != null) {
+            // Upgrade handler do nothing
+          } else {
+            if (version == HttpVersion.HTTP_2 && !client.getOptions().isHttp2ClearTextUpgrade()) {
+              http2Connected(listener, context, ch);
+            } else {
+              http1xConnected(listener, version, host, port, false, context, ch);
+            }
+          }
+        }
+      } else {
+        connectFailed(context, null, listener, res.cause());
+      }
+    };
+
+    channelProvider.connect(client.getVertx(), bootstrap, client.getOptions().getProxyOptions(), SocketAddress.inetSocketAddress(port, host), channelInitializer, channelHandler);
+
+    return weight;
+  }
+
+  private void applyConnectionOptions(Bootstrap bootstrap) {
+    client.getVertx().transport().configure(options, bootstrap);
+  }
+
+  private void applyHttp2ConnectionOptions(ChannelPipeline pipeline) {
+    if (client.getOptions().getIdleTimeout() > 0) {
+      pipeline.addLast("idle", new IdleStateHandler(0, 0, options.getIdleTimeout()));
+    }
+  }
+
+  private void applyHttp1xConnectionOptions(ChannelPipeline pipeline) {
+    if (client.getOptions().getLogActivity()) {
+      pipeline.addLast("logging", new LoggingHandler());
+    }
+    pipeline.addLast("codec", new HttpClientCodec(
+      client.getOptions().getMaxInitialLineLength(),
+      client.getOptions().getMaxHeaderSize(),
+      client.getOptions().getMaxChunkSize(),
+      false,
+      false,
+      client.getOptions().getDecoderInitialBufferSize()));
+    if (client.getOptions().isTryUseCompression()) {
+      pipeline.addLast("inflater", new HttpContentDecompressor(true) {
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+          try {
+            super.channelInactive(ctx);
+          } catch (Exception e) {
+            // Workaround for https://github.com/eclipse/vert.x/issues/2212 until we can get
+            // the proper Netty fix https://github.com/netty/netty/pull/7415
+            // ctx.fireExceptionCaught(e);
+            ctx.fireChannelInactive();
+          }
+        }
+      });
+    }
+    if (client.getOptions().getIdleTimeout() > 0) {
+      pipeline.addLast("idle", new IdleStateHandler(0, 0, client.getOptions().getIdleTimeout()));
+    }
+  }
+
+  private void handshakeFailure(ContextImpl context, Channel ch, Throwable cause, ConnectionListener<HttpClientConnection> listener) {
+    SSLHandshakeException sslException = new SSLHandshakeException("Failed to create SSL connection");
+    if (cause != null) {
+      sslException.initCause(cause);
+    }
+    connectFailed(context, ch, listener, sslException);
+  }
+
+  private void http1xConnected(ConnectionListener<HttpClientConnection> listener,
+                               HttpVersion version,
+                               String host,
+                               int port,
+                               boolean ssl,
+                               ContextImpl context,
+                               Channel ch) {
+    Http1xClientHandler clientHandler = new Http1xClientHandler(
+      listener,
+      context,
+      version,
+      host,
+      port,
+      ssl,
+      client,
+      metric,
+      client.metrics());
+    clientHandler.addHandler(conn -> {
+      listener.onConnectSuccess(conn, http1MaxConcurrency, ch, context, weight, http1Weight);
+    });
+    clientHandler.removeHandler(conn -> {
+      listener.onDiscard();
+    });
+    ch.pipeline().addLast("handler", clientHandler);
+  }
+
+  private void http2Connected(ConnectionListener<HttpClientConnection> listener,
+                              ContextImpl context,
+                              Channel ch) {
+    try {
+      boolean upgrade;
+      upgrade = ch.pipeline().get(SslHandler.class) == null && options.isHttp2ClearTextUpgrade();
+      VertxHttp2ConnectionHandler<Http2ClientConnection> handler = new VertxHttp2ConnectionHandlerBuilder<Http2ClientConnection>(ch)
+        .server(false)
+        .clientUpgrade(upgrade)
+        .useCompression(client.getOptions().isTryUseCompression())
+        .initialSettings(client.getOptions().getInitialSettings())
+        .connectionFactory(connHandler -> new Http2ClientConnection(listener, metric, client, context, connHandler, metrics))
+        .logEnabled(options.getLogActivity())
+        .build();
+      handler.addHandler(conn -> {
+        if (options.getHttp2ConnectionWindowSize() > 0) {
+          conn.setWindowSize(options.getHttp2ConnectionWindowSize());
+        }
+        if (metrics != null) {
+          Object metric = metrics.connected(conn.remoteAddress(), conn.remoteName());
+          conn.metric(metric);
+        }
+        long concurrency = conn.remoteSettings().getMaxConcurrentStreams();
+        if (http2MaxConcurrency > 0) {
+          concurrency = Math.min(concurrency, http2MaxConcurrency);
+        }
+        listener.onConnectSuccess(conn, concurrency, ch, context, weight, http2Weight);
+      });
+      handler.removeHandler(conn -> {
+        if (metrics != null) {
+          metrics.endpointDisconnected(metric, conn.metric());
+        }
+        listener.onDiscard();
+      });
+    } catch (Exception e) {
+      connectFailed(context, ch, listener, e);
+    }
+  }
+
+  private void connectFailed(ContextImpl context, Channel ch, ConnectionListener<HttpClientConnection> listener, Throwable t) {
+    if (ch != null) {
+      try {
+        ch.close();
+      } catch (Exception ignore) {
+      }
+    }
+    listener.onConnectFailure(context, t, weight);
+  }
+}
diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java b/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java
--- a/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java
@@ -16,6 +16,9 @@
 
 package io.vertx.core.http.impl;
 
+import io.netty.channel.Channel;
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Handler;
 import io.vertx.core.http.HttpConnection;
 import io.vertx.core.impl.ContextImpl;
 
@@ -24,18 +27,16 @@
  */
 interface HttpClientConnection extends HttpConnection {
 
-  ContextImpl getContext();
+  Channel channel();
 
   void reportBytesWritten(long numberOfBytes);
 
   void reportBytesRead(long s);
 
-  /**
-   * Check if the connection is valid for creating streams. The connection might be closed or a {@literal GOAWAY}
-   * frame could have been sent or received.
-   */
-  boolean isValid();
-
   void close();
 
+  void createStream(HttpClientRequestImpl req, Handler<AsyncResult<HttpClientStream>> handler);
+
+  ContextImpl getContext();
+
 }
diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java b/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
@@ -16,32 +16,19 @@
 
 package io.vertx.core.http.impl;
 
-import io.vertx.core.Closeable;
-import io.vertx.core.Context;
-import io.vertx.core.Future;
-import io.vertx.core.Handler;
-import io.vertx.core.MultiMap;
-import io.vertx.core.VertxException;
-import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientOptions;
-import io.vertx.core.http.HttpClientRequest;
-import io.vertx.core.http.HttpClientResponse;
-import io.vertx.core.http.HttpHeaders;
-import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.HttpVersion;
-import io.vertx.core.http.RequestOptions;
-import io.vertx.core.http.WebSocket;
-import io.vertx.core.http.WebsocketVersion;
+import io.vertx.core.*;
+import io.vertx.core.http.*;
 import io.vertx.core.impl.ContextImpl;
+import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
 import io.vertx.core.net.ProxyOptions;
 import io.vertx.core.net.ProxyType;
 import io.vertx.core.net.impl.SSLHelper;
+import io.vertx.core.spi.metrics.HttpClientMetrics;
 import io.vertx.core.spi.metrics.Metrics;
 import io.vertx.core.spi.metrics.MetricsProvider;
-import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.core.streams.ReadStream;
 
 import java.net.MalformedURLException;
@@ -53,6 +40,8 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
 import java.util.function.Function;
 
 /**
@@ -107,15 +96,20 @@ public class HttpClientImpl implements HttpClient, MetricsProvider {
   private final VertxInternal vertx;
   private final HttpClientOptions options;
   private final ContextImpl creatingContext;
-  private final ConnectionManager connectionManager;
+  private final ConnectionManager websocketCM; // The queue manager for websockets
+  private final ConnectionManager httpCM; // The queue manager for requests
   private final Closeable closeHook;
   private final ProxyType proxyType;
   private final SSLHelper sslHelper;
+  private final HttpClientMetrics metrics;
+  private final boolean keepAlive;
+  private final boolean pipelining;
   private volatile boolean closed;
   private volatile Function<HttpClientResponse, Future<HttpClientRequest>> redirectHandler = DEFAULT_HANDLER;
 
   public HttpClientImpl(VertxInternal vertx, HttpClientOptions options) {
     this.vertx = vertx;
+    this.metrics = vertx.metricsSPI() != null ? vertx.metricsSPI().createMetrics(this, options) : null;
     this.options = new HttpClientOptions(options);
     List<HttpVersion> alpnVersions = options.getAlpnVersions();
     if (alpnVersions == null || alpnVersions.isEmpty()) {
@@ -128,8 +122,11 @@ public HttpClientImpl(VertxInternal vertx, HttpClientOptions options) {
           break;
       }
     }
+    this.keepAlive = options.isKeepAlive();
+    this.pipelining = options.isPipelining();
     this.sslHelper = new SSLHelper(options, options.getKeyCertOptions(), options.getTrustOptions()).
         setApplicationProtocols(alpnVersions);
+    sslHelper.validate(vertx);
     this.creatingContext = vertx.getContext();
     closeHook = completionHandler -> {
       HttpClientImpl.this.close();
@@ -144,11 +141,19 @@ public HttpClientImpl(VertxInternal vertx, HttpClientOptions options) {
       }
       creatingContext.addCloseHook(closeHook);
     }
-    VertxMetrics metrics = vertx.metricsSPI();
-    connectionManager = new ConnectionManager(this, metrics != null ? metrics.createMetrics(this, options) : null);
+    if (!keepAlive && pipelining) {
+      throw new IllegalStateException("Cannot have pipelining with no keep alive");
+    }
+    long maxWeight = options.getMaxPoolSize() * options.getHttp2MaxPoolSize();
+    websocketCM = new ConnectionManager(this, metrics, HttpVersion.HTTP_1_1, maxWeight, options.getMaxWaitQueueSize());
+    httpCM = new ConnectionManager(this, metrics, options.getProtocolVersion(), maxWeight, options.getMaxWaitQueueSize());
     proxyType = options.getProxyOptions() != null ? options.getProxyOptions().getType() : null;
   }
 
+  HttpClientMetrics metrics() {
+    return metrics;
+  }
+
   @Override
   public HttpClient websocket(RequestOptions options, Handler<WebSocket> wsConnect) {
     return websocket(options, null, wsConnect);
@@ -889,7 +894,11 @@ public synchronized void close() {
     if (creatingContext != null) {
       creatingContext.removeCloseHook(closeHook);
     }
-    connectionManager.close();
+    websocketCM.close();
+    httpCM.close();
+    if (metrics != null) {
+      metrics.close();
+    }
   }
 
   @Override
@@ -899,7 +908,7 @@ public boolean isMetricsEnabled() {
 
   @Override
   public Metrics getMetrics() {
-    return connectionManager.metrics();
+    return metrics;
   }
 
   @Override
@@ -920,34 +929,29 @@ public HttpClientOptions getOptions() {
     return options;
   }
 
-  void getConnectionForWebsocket(boolean ssl,
+  private void getConnectionForWebsocket(boolean ssl,
                                  int port,
                                  String host,
-                                 Handler<ClientConnection> handler,
+                                 Handler<Http1xClientConnection> handler,
                                  Handler<Throwable> connectionExceptionHandler,
                                  ContextImpl context) {
-    connectionManager.getConnectionForWebsocket(ssl, port, host, new Waiter(null, context) {
-      @Override
-      void handleConnection(HttpClientConnection conn) {
-      }
-      @Override
-      void handleStream(HttpClientStream stream) {
-        // Use some variance for this
-        handler.handle((ClientConnection) stream);
-      }
-      @Override
-      void handleFailure(Throwable failure) {
+    websocketCM.getConnection(host, ssl, port, host, null, (ctx, conn) -> {
+      ctx.executeFromIO(() -> {
+        handler.handle((Http1xClientConnection) conn);
+      });
+      return true;
+    }, (ctx, failure) -> {
+      ctx.executeFromIO(() -> {
         connectionExceptionHandler.handle(failure);
-      }
-      @Override
-      boolean isCancelled() {
-        return false;
-      }
+      });
     });
   }
 
-  void getConnectionForRequest(String peerHost, boolean ssl, int port, String host, Waiter waiter) {
-    connectionManager.getConnectionForRequest(options.getProtocolVersion(), peerHost, ssl, port, host, waiter);
+  void getConnectionForRequest(String peerHost, boolean ssl, int port, String host,
+                               Handler<HttpConnection> connectionHandler,
+                               BiFunction<ContextInternal, HttpClientConnection, Boolean> onSuccess,
+                               BiConsumer<ContextInternal, Throwable> onFailure) {
+    httpCM.getConnection(peerHost, ssl, port, host, connectionHandler, onSuccess, onFailure);
   }
 
   /**
@@ -1075,11 +1079,7 @@ public synchronized ReadStream<WebSocket> handler(Handler<WebSocket> handler) {
         }
         getConnectionForWebsocket(ssl != null ? ssl : options.isSsl(), port, host, conn -> {
           conn.exceptionHandler(connectionExceptionHandler);
-          if (conn.isValid()) {
-            conn.toWebSocket(requestURI, headers, version, subProtocols, options.getMaxWebsocketFrameSize(), wsConnect);
-          } else {
-            websocket(port, host, requestURI, headers, version, subProtocols, wsConnect);
-          }
+          conn.toWebSocket(requestURI, headers, version, subProtocols, options.getMaxWebsocketFrameSize(), wsConnect);
         }, connectionExceptionHandler, context);
       }
       return this;
diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientRequestImpl.java b/src/main/java/io/vertx/core/http/impl/HttpClientRequestImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpClientRequestImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpClientRequestImpl.java
@@ -20,9 +20,7 @@
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
 import io.vertx.codegen.annotations.Nullable;
-import io.vertx.core.Future;
-import io.vertx.core.Handler;
-import io.vertx.core.MultiMap;
+import io.vertx.core.*;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.CaseInsensitiveHeaders;
 import io.vertx.core.http.HttpClientRequest;
@@ -31,7 +29,10 @@
 import io.vertx.core.http.HttpFrame;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpVersion;
+import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.impl.VertxInternal;
+import io.vertx.core.logging.Logger;
+import io.vertx.core.logging.LoggerFactory;
 import io.vertx.core.net.NetSocket;
 
 import java.util.List;
@@ -50,6 +51,8 @@
  */
 public class HttpClientRequestImpl extends HttpClientRequestBase implements HttpClientRequest {
 
+  static final Logger log = LoggerFactory.getLogger(ConnectionManager.class);
+
   private final VertxInternal vertx;
   private Handler<HttpClientResponse> respHandler;
   private Handler<Void> endHandler;
@@ -62,20 +65,20 @@ public class HttpClientRequestImpl extends HttpClientRequestBase implements Http
   private Handler<Void> drainHandler;
   private Handler<HttpClientRequest> pushHandler;
   private Handler<HttpConnection> connectionHandler;
-  private boolean headWritten;
+  private Handler<HttpVersion> headersCompletionHandler;
   private boolean completed;
   private Handler<Void> completionHandler;
   private Long reset;
-  private HttpClientResponseImpl response;
   private ByteBuf pendingChunks;
   private CompositeByteBuf cachedChunks;
   private int pendingMaxSize = -1;
   private int followRedirects;
   private boolean connecting;
-  private boolean writeHead;
   private long written;
   private CaseInsensitiveHeaders headers;
 
+  // completed => drainHandler = null
+
   HttpClientRequestImpl(HttpClientImpl client, boolean ssl, HttpMethod method, String host, int port,
                         String relativeURI, VertxInternal vertx) {
     super(client, ssl, method, host, port, relativeURI);
@@ -139,6 +142,10 @@ public HttpClientRequest endHandler(Handler<Void> endHandler) {
 
   @Override
   public HttpClientRequestImpl setChunked(boolean chunked) {
+
+    // If connecting -> write buffer in list
+    // If connected -> check if the buffer needs to be written
+
     synchronized (getLock()) {
       checkComplete();
       if (written > 0) {
@@ -217,36 +224,6 @@ public HttpClientRequest putHeader(String name, Iterable<String> values) {
     }
   }
 
-  @Override
-  public HttpClientRequestImpl write(Buffer chunk) {
-    synchronized (getLock()) {
-      checkComplete();
-      checkResponseHandler();
-      ByteBuf buf = chunk.getByteBuf();
-      write(buf, false);
-      return this;
-    }
-  }
-
-  @Override
-  public HttpClientRequestImpl write(String chunk) {
-    synchronized (getLock()) {
-      checkComplete();
-      checkResponseHandler();
-      return write(Buffer.buffer(chunk));
-    }
-  }
-
-  @Override
-  public HttpClientRequestImpl write(String chunk, String enc) {
-    synchronized (getLock()) {
-      Objects.requireNonNull(enc, "no null encoding accepted");
-      checkComplete();
-      checkResponseHandler();
-      return write(Buffer.buffer(chunk, enc));
-    }
-  }
-
   @Override
   public HttpClientRequest setWriteQueueMaxSize(int maxSize) {
     synchronized (getLock()) {
@@ -312,53 +289,15 @@ public HttpClientRequest sendHead(Handler<HttpVersion> completionHandler) {
       checkComplete();
       checkResponseHandler();
       if (stream != null) {
-        if (!headWritten) {
-          writeHead();
-          if (completionHandler != null) {
-            completionHandler.handle(stream.version());
-          }
-        }
+        throw new IllegalStateException("Head already written");
       } else {
-        connect(completionHandler);
-        writeHead = true;
+        headersCompletionHandler = completionHandler;
+        connect();
       }
       return this;
     }
   }
 
-  @Override
-  public void end(String chunk) {
-    synchronized (getLock()) {
-      end(Buffer.buffer(chunk));
-    }
-  }
-
-  @Override
-  public void end(String chunk, String enc) {
-    synchronized (getLock()) {
-      Objects.requireNonNull(enc, "no null encoding accepted");
-      end(Buffer.buffer(chunk, enc));
-    }
-  }
-
-  @Override
-  public void end(Buffer chunk) {
-    synchronized (getLock()) {
-      checkComplete();
-      checkResponseHandler();
-      write(chunk.getByteBuf(), true);
-    }
-  }
-
-  @Override
-  public void end() {
-    synchronized (getLock()) {
-      checkComplete();
-      checkResponseHandler();
-      write(null, true);
-    }
-  }
-
   @Override
   public HttpClientRequest putHeader(CharSequence name, CharSequence value) {
     synchronized (getLock()) {
@@ -390,8 +329,7 @@ public boolean reset(long code) {
     synchronized (getLock()) {
       if (reset == null) {
         reset = code;
-        if (!completed) {
-          completed = true;
+        if (tryComplete()) {
           if (completionHandler != null) {
             completionHandler.handle(null);
           }
@@ -405,6 +343,16 @@ public boolean reset(long code) {
     }
   }
 
+  private boolean tryComplete() {
+    if (!completed) {
+      completed = true;
+      drainHandler = null;
+      return true;
+    } else {
+      return false;
+    }
+  }
+
   @Override
   public HttpConnection connection() {
     synchronized (getLock()) {
@@ -433,7 +381,7 @@ public HttpClientRequest writeCustomFrame(int type, int flags, Buffer payload) {
 
   void handleDrained() {
     synchronized (getLock()) {
-      if (!completed && drainHandler != null) {
+      if (drainHandler != null) {
         try {
           drainHandler.handle(null);
         } catch (Throwable t) {
@@ -443,7 +391,7 @@ void handleDrained() {
     }
   }
 
-  private void handleNextRequest(HttpClientResponse resp, HttpClientRequestImpl next, long timeoutMs) {
+  private void handleNextRequest(HttpClientRequestImpl next, long timeoutMs) {
     next.handler(respHandler);
     next.exceptionHandler(exceptionHandler());
     exceptionHandler(null);
@@ -507,14 +455,13 @@ else if (completed) {
 
   protected void doHandleResponse(HttpClientResponseImpl resp, long timeoutMs) {
     if (reset == null) {
-      response = resp;
       int statusCode = resp.statusCode();
       if (followRedirects > 0 && statusCode >= 300 && statusCode < 400) {
         Future<HttpClientRequest> next = client.redirectHandler().apply(resp);
         if (next != null) {
           next.setHandler(ar -> {
             if (ar.succeeded()) {
-              handleNextRequest(resp, (HttpClientRequestImpl) ar.result(), timeoutMs);
+              handleNextRequest((HttpClientRequestImpl) ar.result(), timeoutMs);
             } else {
               handleException(ar.cause());
             }
@@ -691,42 +638,13 @@ public HttpClientResponse exceptionHandler(Handler<Throwable> handler) {
     };
   }
 
-  private synchronized void connect(Handler<HttpVersion> headersCompletionHandler) {
+  private synchronized void connect() {
     if (!connecting) {
 
       if (method == HttpMethod.OTHER && rawMethod == null) {
         throw new IllegalStateException("You must provide a rawMethod when using an HttpMethod.OTHER method");
       }
 
-      Waiter waiter = new Waiter(this, vertx.getContext()) {
-
-        @Override
-        void handleFailure(Throwable failure) {
-          handleException(failure);
-        }
-
-        @Override
-        void handleConnection(HttpClientConnection conn) {
-          synchronized (HttpClientRequestImpl.this) {
-            if (connectionHandler != null) {
-              connectionHandler.handle(conn);
-            }
-          }
-        }
-
-        @Override
-        void handleStream(HttpClientStream stream) {
-          connected(stream, headersCompletionHandler);
-        }
-
-        @Override
-        boolean isCancelled() {
-          // No need to synchronize as the thread is the same that set exceptionOccurred to true
-          // exceptionOccurred=true getting the connection => it's a TimeoutException
-          return exceptionOccurred != null || reset != null;
-        }
-      };
-
       String peerHost;
       if (hostHeader != null) {
         int idx = hostHeader.lastIndexOf(':');
@@ -742,18 +660,45 @@ boolean isCancelled() {
       // We defer actual connection until the first part of body is written or end is called
       // This gives the user an opportunity to set an exception handler before connecting so
       // they can capture any exceptions on connection
-      client.getConnectionForRequest(peerHost, ssl, port, host, waiter);
+      client.getConnectionForRequest(peerHost, ssl, port, host, connectionHandler, (ctx, conn) -> {
+        // No need to synchronize as the thread is the same that set exceptionOccurred to true
+        // exceptionOccurred=true getting the connection => it's a TimeoutException
+        if (exceptionOccurred != null || reset != null) {
+          return false;
+        }
+        // checkContext(ctx);
+        conn.createStream(HttpClientRequestImpl.this, ar -> {
+          if (ar.succeeded()) {
+            HttpClientStream stream = ar.result();
+            ctx.executeFromIO(() -> {
+              connected(stream, HttpClientRequestImpl.this.headersCompletionHandler);
+            });
+          } else {
+            throw new RuntimeException(ar.cause());
+          }
+        });
+        return true;
+      }, (ctx, failure) -> {
+        ctx.executeFromIO(() -> {
+          handleException(failure);
+        });
+      });
       connecting = true;
     }
   }
 
+  synchronized void retry() {
+    connecting = false;
+    connect();
+  }
+
   private void connected(HttpClientStream stream, Handler<HttpVersion> headersCompletionHandler) {
 
     HttpClientConnection conn = stream.connection();
 
     synchronized (this) {
       this.stream = stream;
-      stream.beginRequest(this);
+      stream.beginRequest();
 
       // If anything was written or the request ended before we got the connection, then
       // we need to write it now
@@ -768,15 +713,13 @@ private void connected(HttpClientStream stream, Handler<HttpVersion> headersComp
 
         if (completed) {
           // we also need to write the head so optimize this and write all out in once
-          writeHeadWithContent(pending, true);
-
+          stream.writeHeadWithContent(method, rawMethod, uri, headers, hostHeader(), chunked, pending, true);
           conn.reportBytesWritten(written);
-
           if (respHandler != null) {
             this.stream.endRequest();
           }
         } else {
-          writeHeadWithContent(pending, false);
+          stream.writeHeadWithContent(method, rawMethod, uri, headers, hostHeader(), chunked, pending, false);
           if (headersCompletionHandler != null) {
             headersCompletionHandler.handle(stream.version());
           }
@@ -784,19 +727,15 @@ private void connected(HttpClientStream stream, Handler<HttpVersion> headersComp
       } else {
         if (completed) {
           // we also need to write the head so optimize this and write all out in once
-          writeHeadWithContent(null, true);
-
+          stream.writeHeadWithContent(method, rawMethod, uri, headers, hostHeader(), chunked, null, true);
           conn.reportBytesWritten(written);
-
           if (respHandler != null) {
             this.stream.endRequest();
           }
         } else {
-          if (writeHead) {
-            writeHead();
-            if (headersCompletionHandler != null) {
-              headersCompletionHandler.handle(stream.version());
-            }
+          stream.writeHead(method, rawMethod, uri, headers, hostHeader(), chunked);
+          if (headersCompletionHandler != null) {
+            headersCompletionHandler.handle(stream.version());
           }
         }
       }
@@ -811,17 +750,54 @@ private boolean contentLengthSet() {
     return headers != null && headers().contains(CONTENT_LENGTH);
   }
 
-  private void writeHead() {
-    stream.writeHead(method, rawMethod, uri, headers, hostHeader(), chunked);
-    headWritten = true;
+  @Override
+  public void end(String chunk) {
+    end(Buffer.buffer(chunk));
+  }
+
+  @Override
+  public void end(String chunk, String enc) {
+    Objects.requireNonNull(enc, "no null encoding accepted");
+    end(Buffer.buffer(chunk, enc));
+  }
+
+  @Override
+  public void end(Buffer chunk) {
+    write(chunk.getByteBuf(), true);
+  }
+
+  @Override
+  public void end() {
+    write(null, true);
+  }
+
+  @Override
+  public HttpClientRequestImpl write(Buffer chunk) {
+    ByteBuf buf = chunk.getByteBuf();
+    write(buf, false);
+    return this;
+  }
+
+  @Override
+  public HttpClientRequestImpl write(String chunk) {
+    return write(Buffer.buffer(chunk));
   }
 
-  private void writeHeadWithContent(ByteBuf buf, boolean end) {
-    stream.writeHeadWithContent(method, rawMethod, uri, headers, hostHeader(), chunked, buf, end);
-    headWritten = true;
+  @Override
+  public HttpClientRequestImpl write(String chunk, String enc) {
+    Objects.requireNonNull(enc, "no null encoding accepted");
+    return write(Buffer.buffer(chunk, enc));
   }
 
   private void write(ByteBuf buff, boolean end) {
+    synchronized (getLock()) {
+      checkComplete();
+      checkResponseHandler();
+      _write(buff, end);
+    }
+  }
+
+  private void _write(ByteBuf buff, boolean end) {
     if (buff == null && !end) {
       // nothing to write to the connection just return
       return;
@@ -864,33 +840,26 @@ private void write(ByteBuf buff, boolean end) {
           pending.addComponent(true, buff);
         }
       }
-      connect(null);
-    } else {
-      if (!headWritten) {
-        writeHeadWithContent(buff, end);
-      } else {
-        stream.writeBuffer(buff, end);
+      if (end) {
+        tryComplete();
+        if (completionHandler != null) {
+          completionHandler.handle(null);
+        }
       }
+      connect();
+    } else {
+      stream.writeBuffer(buff, end);
       if (end) {
         stream.connection().reportBytesWritten(written);
         if (respHandler != null) {
           stream.endRequest();
         }
+        tryComplete();
+        if (completionHandler != null) {
+          completionHandler.handle(null);
+        }
       }
     }
-
-    if (end) {
-      completed = true;
-      if (completionHandler != null) {
-        completionHandler.handle(null);
-      }
-    }
-  }
-
-  void handleResponseEnd() {
-    synchronized (getLock()) {
-      response = null;
-    }
   }
 
   protected void checkComplete() {
diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientStream.java b/src/main/java/io/vertx/core/http/impl/HttpClientStream.java
--- a/src/main/java/io/vertx/core/http/impl/HttpClientStream.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpClientStream.java
@@ -54,7 +54,7 @@ interface HttpClientStream {
 
   void reset(long code);
 
-  void beginRequest(HttpClientRequestImpl request);
+  void beginRequest();
   void endRequest();
 
   NetSocket createNetSocket();
diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java
@@ -122,7 +122,7 @@ public class HttpServerImpl implements HttpServer, Closeable, MetricsProvider {
   private final VertxInternal vertx;
   private final SSLHelper sslHelper;
   private final ContextImpl creatingContext;
-  private final Map<Channel, ServerConnection> connectionMap = new ConcurrentHashMap<>();
+  private final Map<Channel, Http1xServerConnection> connectionMap = new ConcurrentHashMap<>();
   private final Map<Channel, Http2ServerConnection> connectionMap2 = new ConcurrentHashMap<>();
   private final VertxEventLoopGroup availableWorkers = new VertxEventLoopGroup();
   private final HandlerManager<HttpHandlers> httpHandlerMgr = new HandlerManager<>(availableWorkers);
@@ -395,26 +395,37 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
   }
 
   private VertxHttp2ConnectionHandler<Http2ServerConnection> setHandler(HandlerHolder<HttpHandlers> holder, Http2Settings upgrade, Channel ch) {
-    return new VertxHttp2ConnectionHandlerBuilder<Http2ServerConnection>(ch)
-        .connectionMap(connectionMap2)
-        .server(true)
-        .serverUpgrade(upgrade)
-        .useCompression(options.isCompressionSupported())
-        .useDecompression(options.isDecompressionSupported())
-        .compressionLevel(options.getCompressionLevel())
-        .initialSettings(options.getInitialSettings())
-        .connectionFactory(connHandler -> {
-          Http2ServerConnection conn = new Http2ServerConnection(holder.context, serverOrigin, connHandler, options, holder.handler.requesthHandler, metrics);
-          if (metrics != null) {
-            conn.metric(metrics.connected(conn.remoteAddress(), conn.remoteName()));
-          }
-          if (options.getHttp2ConnectionWindowSize() > 0) {
-            conn.setWindowSize(options.getHttp2ConnectionWindowSize());
-          }
-          return conn;
-        })
-        .logEnabled(logEnabled)
-        .build();
+    VertxHttp2ConnectionHandler<Http2ServerConnection> handler = new VertxHttp2ConnectionHandlerBuilder<Http2ServerConnection>(ch)
+      .server(true)
+      .serverUpgrade(upgrade)
+      .useCompression(options.isCompressionSupported())
+      .useDecompression(options.isDecompressionSupported())
+      .compressionLevel(options.getCompressionLevel())
+      .initialSettings(options.getInitialSettings())
+      .connectionFactory(connHandler -> {
+        Http2ServerConnection conn = new Http2ServerConnection(holder.context, serverOrigin, connHandler, options, holder.handler.requesthHandler, metrics);
+        if (metrics != null) {
+          conn.metric(metrics.connected(conn.remoteAddress(), conn.remoteName()));
+        }
+        if (options.getHttp2ConnectionWindowSize() > 0) {
+          conn.setWindowSize(options.getHttp2ConnectionWindowSize());
+        }
+        return conn;
+      })
+      .logEnabled(logEnabled)
+      .build();
+    handler.addHandler(conn -> {
+      connectionMap2.put(conn.channel(), conn);
+      if (holder.handler.connectionHandler != null) {
+        holder.context.executeFromIO(() -> {
+          holder.handler.connectionHandler.handle(conn);
+        });
+      }
+    });
+    handler.removeHandler(conn -> {
+      connectionMap2.remove(conn.channel());
+    });
+    return handler;
   }
 
   private void configureHttp1(ChannelPipeline pipeline) {
@@ -444,11 +455,11 @@ private void configureHttp1(ChannelPipeline pipeline) {
       pipeline.addLast("h2c", new Http2UpgradeHandler());
     }
     HandlerHolder<HttpHandlers> holder = httpHandlerMgr.chooseHandler(pipeline.channel().eventLoop());
-    ServerHandler handler;
+    Http1xServerHandler handler;
     if (DISABLE_WEBSOCKETS) {
       // As a performance optimisation you can set a system property to disable websockets altogether which avoids
       // some casting and a header check
-      handler = new ServerHandler(sslHelper, options, serverOrigin, holder, metrics);
+      handler = new Http1xServerHandler(sslHelper, options, serverOrigin, holder, metrics);
     } else {
       handler = new ServerHandlerWithWebSockets(sslHelper, options, serverOrigin, holder, metrics);
     }
@@ -464,12 +475,7 @@ private void configureHttp1(ChannelPipeline pipeline) {
   public void handleHttp2(Channel ch) {
     HandlerHolder<HttpHandlers> holder = httpHandlerMgr.chooseHandler(ch.eventLoop());
     configureHttp2(ch.pipeline());
-    VertxHttp2ConnectionHandler<Http2ServerConnection> handler = setHandler(holder, null, ch);
-    if (holder.handler.connectionHandler != null) {
-      holder.context.executeFromIO(() -> {
-        holder.handler.connectionHandler.handle(handler.connection);
-      });
-    }
+    setHandler(holder, null, ch);
   }
 
   public void configureHttp2(ChannelPipeline pipeline) {
@@ -579,7 +585,7 @@ private void actualClose(final ContextImpl closeContext, final Handler<AsyncResu
 
     ContextImpl currCon = vertx.getContext();
 
-    for (ServerConnection conn : connectionMap.values()) {
+    for (Http1xServerConnection conn : connectionMap.values()) {
       conn.close();
     }
     for (Http2ServerConnection conn : connectionMap2.values()) {
@@ -611,7 +617,7 @@ private void executeCloseDone(final ContextImpl closeContext, final Handler<Asyn
     }
   }
 
-  public class ServerHandlerWithWebSockets extends ServerHandler {
+  public class ServerHandlerWithWebSockets extends Http1xServerHandler {
 
     private boolean closeFrameSent;
     private FullHttpRequest wsRequest;
@@ -623,7 +629,7 @@ public ServerHandlerWithWebSockets(SSLHelper sslHelper, HttpServerOptions option
     }
 
     @Override
-    protected void handleMessage(ServerConnection conn, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
+    protected void handleMessage(Http1xServerConnection conn, ContextImpl context, ChannelHandlerContext chctx, Object msg) throws Exception {
       Channel ch = chctx.channel();
       if (msg instanceof HttpRequest) {
         final HttpRequest request = (HttpRequest) msg;
@@ -708,7 +714,7 @@ protected void handleMessage(ServerConnection conn, ContextImpl context, Channel
       }
     }
 
-    protected void handshake(ServerConnection conn, FullHttpRequest request, Channel ch, ChannelHandlerContext ctx) throws Exception {
+    protected void handshake(Http1xServerConnection conn, FullHttpRequest request, Channel ch, ChannelHandlerContext ctx) throws Exception {
 
       WebSocketServerHandshaker shake = createHandshaker(conn, ch, request);
       if (shake == null) {
diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java
@@ -64,7 +64,7 @@ public class HttpServerRequestImpl implements HttpServerRequest {
 
   private static final Logger log = LoggerFactory.getLogger(HttpServerRequestImpl.class);
 
-  private final ServerConnection conn;
+  private final Http1xServerConnection conn;
   private final HttpRequest request;
   private final HttpServerResponse response;
 
@@ -91,7 +91,7 @@ public class HttpServerRequestImpl implements HttpServerRequest {
   private boolean ended;
 
 
-  HttpServerRequestImpl(ServerConnection conn,
+  HttpServerRequestImpl(Http1xServerConnection conn,
                         HttpRequest request,
                         HttpServerResponse response) {
     this.conn = conn;
diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java
@@ -59,7 +59,7 @@ public class HttpServerResponseImpl implements HttpServerResponse {
   private static final Logger log = LoggerFactory.getLogger(HttpServerResponseImpl.class);
 
   private final VertxInternal vertx;
-  private final ServerConnection conn;
+  private final Http1xServerConnection conn;
   private HttpResponseStatus status;
   private final HttpVersion version;
   private final boolean keepAlive;
@@ -81,7 +81,7 @@ public class HttpServerResponseImpl implements HttpServerResponse {
   private String statusMessage;
   private long bytesWritten;
 
-  HttpServerResponseImpl(final VertxInternal vertx, ServerConnection conn, HttpRequest request) {
+  HttpServerResponseImpl(final VertxInternal vertx, Http1xServerConnection conn, HttpRequest request) {
     this.vertx = vertx;
     this.conn = conn;
     this.version = request.getProtocolVersion();
diff --git a/src/main/java/io/vertx/core/http/impl/HttpUtils.java b/src/main/java/io/vertx/core/http/impl/HttpUtils.java
--- a/src/main/java/io/vertx/core/http/impl/HttpUtils.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpUtils.java
@@ -457,6 +457,16 @@ static HttpVersion toNettyHttpVersion(io.vertx.core.http.HttpVersion version) {
     }
   }
 
+  static io.vertx.core.http.HttpVersion toVertxHttpVersion(HttpVersion version) {
+    if (version == io.netty.handler.codec.http.HttpVersion.HTTP_1_0) {
+      return io.vertx.core.http.HttpVersion.HTTP_1_0;
+    } else if (version == io.netty.handler.codec.http.HttpVersion.HTTP_1_1) {
+      return io.vertx.core.http.HttpVersion.HTTP_1_1;
+    } else {
+      return null;
+    }
+  }
+
   static io.vertx.core.http.HttpMethod toVertxMethod(String method) {
     try {
       return io.vertx.core.http.HttpMethod.valueOf(method);
diff --git a/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandler.java b/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandler.java
--- a/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandler.java
+++ b/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandler.java
@@ -17,20 +17,10 @@
 package io.vertx.core.http.impl;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
-import io.netty.handler.codec.http2.Http2Connection;
-import io.netty.handler.codec.http2.Http2ConnectionDecoder;
-import io.netty.handler.codec.http2.Http2ConnectionEncoder;
-import io.netty.handler.codec.http2.Http2ConnectionHandler;
-import io.netty.handler.codec.http2.Http2Exception;
-import io.netty.handler.codec.http2.Http2Flags;
-import io.netty.handler.codec.http2.Http2Headers;
-import io.netty.handler.codec.http2.Http2RemoteFlowController;
-import io.netty.handler.codec.http2.Http2Settings;
-import io.netty.handler.codec.http2.Http2Stream;
+import io.netty.handler.codec.http2.*;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 import io.netty.util.concurrent.EventExecutor;
@@ -38,24 +28,29 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 
-import java.util.Map;
+import java.util.function.Function;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-class VertxHttp2ConnectionHandler<C extends Http2ConnectionBase> extends Http2ConnectionHandler implements Http2Connection.Listener {
+class VertxHttp2ConnectionHandler<C extends Http2ConnectionBase> extends Http2ConnectionHandler implements Http2FrameListener, Http2Connection.Listener {
 
-  private final Map<Channel, ? super C> connectionMap;
-  C connection;
+  private final Function<VertxHttp2ConnectionHandler<C>, C> connectionFactory;
+  private C connection;
   private ChannelHandlerContext chctx;
+  private Handler<C> addHandler;
+  private Handler<C> removeHandler;
+  private final boolean useDecompressor;
 
   public VertxHttp2ConnectionHandler(
-      Map<Channel, ? super C> connectionMap,
+      Function<VertxHttp2ConnectionHandler<C>, C> connectionFactory,
+      boolean useDecompressor,
       Http2ConnectionDecoder decoder,
       Http2ConnectionEncoder encoder,
       Http2Settings initialSettings) {
     super(decoder, encoder, initialSettings);
-    this.connectionMap = connectionMap;
+    this.connectionFactory = connectionFactory;
+    this.useDecompressor = useDecompressor;
     encoder().flowController().listener(s -> {
       if (connection != null) {
         connection.onStreamwritabilityChanged(s);
@@ -68,17 +63,34 @@ public ChannelHandlerContext context() {
     return chctx;
   }
 
+  /**
+   * Set an handler to be called when the connection is set on this handler.
+   *
+   * @param handler the handler to be notified
+   * @return this
+   */
+  public VertxHttp2ConnectionHandler<C> addHandler(Handler<C> handler) {
+    this.addHandler = handler;
+    return this;
+  }
+
+  /**
+   * Set an handler to be called when the connection is unset from this handler.
+   *
+   * @param handler the handler to be notified
+   * @return this
+   */
+  public VertxHttp2ConnectionHandler<C> removeHandler(Handler<C> handler) {
+    this.removeHandler = handler;
+    return this;
+  }
+
   @Override
   public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
     super.handlerAdded(ctx);
     chctx = ctx;
   }
 
-  void init(C conn) {
-    connection = conn;
-    connectionMap.put(chctx.channel(), connection);
-  }
-
   @Override
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
     super.exceptionCaught(ctx, cause);
@@ -88,13 +100,20 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
   @Override
   public void channelActive(ChannelHandlerContext ctx) throws Exception {
     super.channelActive(ctx);
+
+    // super call writes the connection preface
+    // we need to flush to send it
+    // this is called only on the client
+    ctx.flush();
   }
 
   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
     super.channelInactive(ctx);
-    connectionMap.remove(ctx.channel());
     connection.getContext().executeFromIO(connection::handleClosed);
+    if (removeHandler != null) {
+      removeHandler.handle(connection);
+    }
   }
 
   @Override
@@ -321,4 +340,81 @@ void writePushPromise(int streamId, Http2Headers headers, Handler<AsyncResult<In
   private void _writePushPromise(int streamId, int promisedStreamId, Http2Headers headers, ChannelPromise promise) {
     encoder().writePushPromise(chctx, streamId, promisedStreamId, headers, 0, promise);
   }
+
+  // Http2FrameListener
+
+  @Override
+  public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding, boolean endOfStream) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency, short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
+    assert connection != null;
+    connection.onHeadersRead(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endOfStream);
+  }
+
+  @Override
+  public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight, boolean exclusive) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
+    connection = connectionFactory.apply(this);
+    if (useDecompressor) {
+      decoder().frameListener(new DelegatingDecompressorFrameListener(decoder().connection(), connection));
+    } else {
+      decoder().frameListener(connection);
+    }
+    connection.onSettingsRead(ctx, settings);
+    if (addHandler != null) {
+      addHandler.handle(connection);
+    }
+  }
+
+  @Override
+  public void onPingRead(ChannelHandlerContext ctx, ByteBuf data) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onPingAckRead(ChannelHandlerContext ctx, ByteBuf data) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId, Http2Headers headers, int padding) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags, ByteBuf payload) throws Http2Exception {
+    throw new UnsupportedOperationException();
+  }
 }
diff --git a/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandlerBuilder.java b/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandlerBuilder.java
--- a/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandlerBuilder.java
+++ b/src/main/java/io/vertx/core/http/impl/VertxHttp2ConnectionHandlerBuilder.java
@@ -21,7 +21,6 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http2.AbstractHttp2ConnectionHandlerBuilder;
 import io.netty.handler.codec.http2.CompressorHttp2ConnectionEncoder;
-import io.netty.handler.codec.http2.DelegatingDecompressorFrameListener;
 import io.netty.handler.codec.http2.Http2ConnectionDecoder;
 import io.netty.handler.codec.http2.Http2ConnectionEncoder;
 import io.netty.handler.codec.http2.Http2Exception;
@@ -33,7 +32,6 @@
 import io.netty.handler.logging.LogLevel;
 import io.vertx.core.http.HttpServerOptions;
 
-import java.util.Map;
 import java.util.function.Function;
 
 /**
@@ -44,7 +42,6 @@
 class VertxHttp2ConnectionHandlerBuilder<C extends Http2ConnectionBase> extends AbstractHttp2ConnectionHandlerBuilder<VertxHttp2ConnectionHandler<C>, VertxHttp2ConnectionHandlerBuilder<C>> {
 
   private final Channel channel;
-  private Map<Channel, ? super C> connectionMap;
   private boolean useCompression;
   private boolean useDecompression;
   private int compressionLevel = HttpServerOptions.DEFAULT_COMPRESSION_LEVEL;
@@ -62,11 +59,6 @@ protected VertxHttp2ConnectionHandlerBuilder<C> server(boolean isServer) {
     return super.server(isServer);
   }
 
-  VertxHttp2ConnectionHandlerBuilder<C> connectionMap(Map<Channel, ? super C> connectionMap) {
-    this.connectionMap = connectionMap;
-    return this;
-  }
-
   VertxHttp2ConnectionHandlerBuilder<C> initialSettings(io.vertx.core.http.Http2Settings settings) {
     this.initialSettings = settings;
     return this;
@@ -88,22 +80,22 @@ public VertxHttp2ConnectionHandlerBuilder<C> serverUpgrade(Http2Settings upgrade
   }
 
   /**
-   * This method allows to set the compression level to be used in the http/2 connection encoder 
-   * (for data sent to client) when compression support is turned on (@see useCompression) and 
+   * This method allows to set the compression level to be used in the http/2 connection encoder
+   * (for data sent to client) when compression support is turned on (@see useCompression) and
    * the client advertises to support deflate/gizip compression in the Accept-Encoding header
-   * 
+   *
    * default value is : 6 (Netty legacy)
-   * 
-   * While one can think that best value is always the maximum compression ratio, 
-   * there's a trade-off to consider: the most compressed level requires the most computatinal work to compress/decompress, 
-   * E.g. you have it set fairly high on a high-volume website, you may experience performance degradation 
-   * and latency on resource serving due to CPU overload, and however - as the comptational work is required also client side 
+   *
+   * While one can think that best value is always the maximum compression ratio,
+   * there's a trade-off to consider: the most compressed level requires the most computatinal work to compress/decompress,
+   * E.g. you have it set fairly high on a high-volume website, you may experience performance degradation
+   * and latency on resource serving due to CPU overload, and however - as the comptational work is required also client side
    * while decompressing - setting an higher compression level can result in an overall higher page load time
    * especially nowadays when many clients are handled mobile devices with a low CPU profile.
-   * 
+   *
    * see also: http://www.gzip.org/algorithm.txt
-   * 
-   * @param compressionLevel integer 1-9, 1 means use fastest algorithm, 9 slower algorithm but better compression ratio 
+   *
+   * @param compressionLevel integer 1-9, 1 means use fastest algorithm, 9 slower algorithm but better compression ratio
    * @return a reference to this instance for fulent API coding style
    */
   VertxHttp2ConnectionHandlerBuilder<C> compressionLevel(int compressionLevel) {
@@ -115,7 +107,7 @@ VertxHttp2ConnectionHandlerBuilder<C> useDecompression(boolean useDecompression)
     this.useDecompression = useDecompression;
     return this;
   }
-  
+
   VertxHttp2ConnectionHandlerBuilder<C> connectionFactory(Function<VertxHttp2ConnectionHandler<C>, C> connectionFactory) {
     this.connectionFactory = connectionFactory;
     return this;
@@ -198,30 +190,20 @@ protected VertxHttp2ConnectionHandler<C> build(Http2ConnectionDecoder decoder, H
       if (useCompression) {
         encoder = new CompressorHttp2ConnectionEncoder(encoder,compressionLevel,CompressorHttp2ConnectionEncoder.DEFAULT_WINDOW_BITS,CompressorHttp2ConnectionEncoder.DEFAULT_MEM_LEVEL);
       }
-      VertxHttp2ConnectionHandler<C> handler = new VertxHttp2ConnectionHandler<>(connectionMap, decoder, encoder, initialSettings);
+      VertxHttp2ConnectionHandler<C> handler = new VertxHttp2ConnectionHandler<>(connectionFactory, useDecompression, decoder, encoder, initialSettings);
       if (serverUpgrade != null) {
         handler.onHttpServerUpgrade(serverUpgrade);
       }
       channel.pipeline().addLast(handler);
-      handler.init(connectionFactory.apply(handler));
-      if (useDecompression) {
-        decoder.frameListener(new DelegatingDecompressorFrameListener(decoder.connection(), handler.connection));
-      } else {
-        decoder.frameListener(handler.connection);
-      }
+      decoder.frameListener(handler);
       return handler;
     } else {
-      VertxHttp2ConnectionHandler<C> handler = new VertxHttp2ConnectionHandler<>(connectionMap, decoder, encoder, initialSettings);
+      VertxHttp2ConnectionHandler<C> handler = new VertxHttp2ConnectionHandler<>(connectionFactory, useCompression, decoder, encoder, initialSettings);
       if (clientUpgrade) {
         handler.onHttpClientUpgrade();
       }
       channel.pipeline().addLast(handler);
-      handler.init(connectionFactory.apply(handler));
-      if (useCompression) {
-        decoder.frameListener(new DelegatingDecompressorFrameListener(decoder.connection(), handler.connection));
-      } else {
-        decoder.frameListener(handler.connection);
-      }
+      decoder.frameListener(handler);
       return handler;
     }
   }
diff --git a/src/main/java/io/vertx/core/http/impl/WebSocketImpl.java b/src/main/java/io/vertx/core/http/impl/WebSocketImpl.java
--- a/src/main/java/io/vertx/core/http/impl/WebSocketImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/WebSocketImpl.java
@@ -17,7 +17,6 @@
 package io.vertx.core.http.impl;
 
 import io.vertx.core.http.WebSocket;
-import io.vertx.core.http.WebSocketFrame;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
 
@@ -33,7 +32,7 @@
 public class WebSocketImpl extends WebSocketImplBase<WebSocket> implements WebSocket {
 
   public WebSocketImpl(VertxInternal vertx,
-                       ClientConnection conn, boolean supportsContinuation,
+                       Http1xClientConnection conn, boolean supportsContinuation,
                        int maxWebSocketFrameSize, int maxWebSocketMessageSize) {
     super(vertx, conn, supportsContinuation, maxWebSocketFrameSize, maxWebSocketMessageSize);
   }
@@ -41,7 +40,7 @@ public WebSocketImpl(VertxInternal vertx,
   @Override
   void handleClosed() {
     synchronized (conn) {
-      HttpClientMetrics metrics = ((ClientConnection) conn).metrics();
+      HttpClientMetrics metrics = ((Http1xClientConnection) conn).metrics();
       if (metrics != null) {
         metrics.disconnected(getMetric());
       }
diff --git a/src/main/java/io/vertx/core/http/impl/pool/ConnectionListener.java b/src/main/java/io/vertx/core/http/impl/pool/ConnectionListener.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/http/impl/pool/ConnectionListener.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright (c) 2011-2013 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ *     The Eclipse Public License is available at
+ *     http://www.eclipse.org/legal/epl-v10.html
+ *
+ *     The Apache License v2.0 is available at
+ *     http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.vertx.core.http.impl.pool;
+
+import io.netty.channel.Channel;
+import io.vertx.core.impl.ContextImpl;
+
+/**
+ * The listener definest the contract used by the {@link ConnectionProvider} to interact with the
+ * connection pool. Its purpose is also to use a connection implementation without a pool.
+ */
+public interface ConnectionListener<C> {
+
+  /**
+   * Callback to signal the connection succeeded and provide all the info requires to manage the connection
+   *
+   * @param conn the connection
+   * @param concurrency the connection concurrency
+   * @param channel the channel
+   * @param context the context
+   * @param initialWeight the initial weight
+   * @param actualWeight the actual weight
+   */
+  void onConnectSuccess(C conn,
+                        long concurrency,
+                        Channel channel,
+                        ContextImpl context,
+                        long initialWeight,
+                        long actualWeight);
+
+  /**
+   * Callback to signal the connection failed.
+   *
+   * @param context the context
+   * @param err the error
+   * @param weight the weight
+   */
+  void onConnectFailure(ContextImpl context, Throwable err, long weight);
+
+  /**
+   * Signals the connrection changed to the {@code concurrency} value.
+   *
+   * @param concurrency the concurrency
+   */
+  void onConcurrencyChange(long concurrency);
+
+  /**
+   * Signals the connection can recycled, it must not give back more than it borrowed.
+   *
+   * @param capacity the capacity to recycle
+   * @param disposable wether the connection can be disposed
+   */
+  void onRecycle(int capacity, boolean disposable);
+
+  /**
+   * Signals the connection must not be used anymore by the pool.
+   */
+  void onDiscard();
+
+}
diff --git a/src/main/java/io/vertx/core/http/impl/pool/ConnectionProvider.java b/src/main/java/io/vertx/core/http/impl/pool/ConnectionProvider.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/http/impl/pool/ConnectionProvider.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2011-2013 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ *     The Eclipse Public License is available at
+ *     http://www.eclipse.org/legal/epl-v10.html
+ *
+ *     The Apache License v2.0 is available at
+ *     http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.vertx.core.http.impl.pool;
+
+import io.vertx.core.impl.ContextImpl;
+
+/**
+ * Provides how the connection manager interacts its connections.
+ *
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ */
+public interface ConnectionProvider<C> {
+
+  /**
+   * Connect to the server and signals the {@code listener} the success with {@link ConnectionListener#onConnectSuccess}
+   * or the failure with {@link ConnectionListener#onConnectFailure}.
+   *
+   * @param listener the listener
+   * @param context the context to use for the connection
+   * @return the initial weight of the connection, which will eventually be corrected when calling the listener
+   */
+  long connect(ConnectionListener<C> listener, ContextImpl context);
+
+  /**
+   * Close a connection.
+   *
+   * @param conn the connection
+   */
+  void close(C conn);
+
+}
diff --git a/src/main/java/io/vertx/core/http/impl/pool/Pool.java b/src/main/java/io/vertx/core/http/impl/pool/Pool.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/http/impl/pool/Pool.java
@@ -0,0 +1,366 @@
+/*
+ * Copyright (c) 2011-2014 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ *     The Eclipse Public License is available at
+ *     http://www.eclipse.org/legal/epl-v10.html
+ *
+ *     The Apache License v2.0 is available at
+ *     http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.vertx.core.http.impl.pool;
+
+import io.netty.channel.Channel;
+import io.vertx.core.Handler;
+import io.vertx.core.http.ConnectionPoolTooBusyException;
+import io.vertx.core.impl.ContextImpl;
+import io.vertx.core.logging.Logger;
+import io.vertx.core.logging.LoggerFactory;
+
+import java.util.*;
+import java.util.function.BiConsumer;
+
+/**
+ * The pool is a queue of waiters and a list of connections.
+ *
+ * Pool invariants:
+ * - a connection in the {@link #available} list has its {@code Holder#capacity > 0}
+ * - the {@link #weight} is the sum of all inflight connections {@link Holder#weight}
+ *
+ * A connection is delivered to a {@link Waiter} on the connection's event loop thread, the waiter must take care of
+ * calling {@link io.vertx.core.impl.ContextInternal#executeFromIO} if necessary.
+ *
+ * Calls to the pool are synchronized on the pool to avoid race conditions and maintain its invariants. This pool can
+ * be called from different threads safely (although it is not encouraged for performance reasons, we benefit from biased
+ * locking which makes the overhead of synchronized near zero), since it synchronizes on the pool.
+ *
+ * In order to avoid deadlocks, acquisition events (success or failure) are dispatched on the event loop thread of the
+ * connection without holding the pool lock.
+ *
+ * To constrain the number of connections the pool maintains a {@link #weight} value that must remain below the the
+ * {@link #maxWeight} value to create a connection. Weight is used instead of counting connection because this pool
+ * can mix connections with different concurrency (HTTP/1 and HTTP/2) and this flexibility is necessary.
+ *
+ * When a connection is created an initial weight is returned by the {@link ConnectionProvider#connect} method and is
+ * added to the current weight. When the channel is connected the {@link ConnectionListener#onConnectSuccess} callback
+ * provides the initial weight returned by the connect method and the actual connection weight so it can be used to
+ * correct the current weight. When the channel fails to connect the {@link ConnectionListener#onConnectFailure} failure
+ * provides the initial weight so it can be used to correct the current weight.
+ *
+ * When a connection is recycled and reaches its full capacity (i.e {@code Holder#concurrency == Holder#capacity},
+ * the behavior depends on the {@link ConnectionListener#onRecycle(int, boolean)} event that release this connection.
+ * When {@code disposable} is {@code true} the connection is closed, otherwise it is maintained in the pool, letting
+ * the borrower define the behavior. HTTP/1 will close the connection and HTTP/2 will maintain it.
+ *
+ * When a waiter asks for a connection, it is either added to the queue (when it's not empty) or attempted to be
+ * served (from the pool or by creating a new connection) or failed. The {@link #waitersCount} is the number
+ * of total waiters (the waiters in {@link #waitersQueue} but also the inflight) so we know if we can close the pool
+ * or not. The {@link #waitersCount} is incremented when a waiter wants to acquire a connection succesfully (i.e
+ * it is either added to the queue or served from the pool) and decremented when the it gets a reply (either with
+ * a connection or with a failure).
+ *
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ */
+public class Pool<C> {
+
+  /**
+   * Pool state associated with a connection.
+   *
+   * @param <C> the connection type
+   */
+  public static class Holder<C> {
+
+    boolean removed;     // Removed
+    C connection;        // The connection instance
+    long concurrency;    // How many times we can borrow from the connection
+    long capacity;       // How many times the connection is currently borrowed (0 <= capacity <= concurrency)
+    Channel channel;     // Transport channel
+    ContextImpl context; // Context associated with the connection
+    long weight;         // The weight that participates in the pool weight
+
+  }
+  private static final Logger log = LoggerFactory.getLogger(Pool.class);
+
+  private final ConnectionProvider<C> connector;
+  private final BiConsumer<Channel, C> connectionAdded;
+  private final BiConsumer<Channel, C> connectionRemoved;
+
+  private final int queueMaxSize;                                   // the queue max size (does not include inflight waiters)
+  private final Queue<Waiter<C>> waitersQueue = new ArrayDeque<>(); // The waiters pending
+  private int waitersCount;                                         // The number of waiters (including the inflight waiters not in the queue)
+
+  private final Deque<Holder<C>> available;                         // Available connections
+
+  private final long maxWeight;                                     // The max weight (equivalent to max pool size)
+  private long weight;                                              // The actual pool weight (equivalent to connection count)
+
+  private boolean closed;
+  private final Handler<Void> poolClosed;
+
+  public Pool(ConnectionProvider<C> connector,
+              int queueMaxSize,
+              long maxWeight,
+              Handler<Void> poolClosed,
+              BiConsumer<Channel, C> connectionAdded,
+              BiConsumer<Channel, C> connectionRemoved) {
+    this.maxWeight = maxWeight;
+    this.connector = connector;
+    this.queueMaxSize = queueMaxSize;
+    this.poolClosed = poolClosed;
+    this.available = new ArrayDeque<>();
+    this.connectionAdded = connectionAdded;
+    this.connectionRemoved = connectionRemoved;
+  }
+
+  public synchronized int waitersInQueue() {
+    return waitersQueue.size();
+  }
+
+  public synchronized int waitersCount() {
+    return waitersCount;
+  }
+
+  public synchronized long weight() {
+    return weight;
+  }
+
+  public synchronized long capacity() {
+    return available.stream().mapToLong(c -> c.capacity).sum();
+  }
+
+  /**
+   * Get a connection for a waiter asynchronously.
+   *
+   * @param waiter the waiter
+   * @return whether the pool can satisfy the request
+   */
+  public synchronized boolean getConnection(Waiter<C> waiter) {
+    if (closed) {
+      return false;
+    }
+    int size = waitersQueue.size();
+    if (size == 0 && acquireConnection(waiter)) {
+      waitersCount++;
+    } else if (queueMaxSize < 0  || size < queueMaxSize) {
+      waitersCount++;
+      waitersQueue.add(waiter);
+    } else {
+      waiter.context.nettyEventLoop().execute(() -> {
+        waiter.handleFailure(waiter.context, new ConnectionPoolTooBusyException("Connection pool reached max wait queue size of " + queueMaxSize));
+      });
+    }
+    return true;
+  }
+
+  /**
+   * Attempt to acquire a connection for the waiter, either borrowed from the pool or by creating a new connection.
+   *
+   * This method does not modify the waitersQueue list.
+   *
+   * @return wether the waiter is assigned a connection (or a future connection)
+   */
+  private boolean acquireConnection(Waiter<C> waiter) {
+    if (available.size() > 0) {
+      Holder<C> conn = available.peek();
+      if (--conn.capacity == 0) {
+        available.poll();
+      }
+      ContextImpl ctx = conn.context;
+      ctx.nettyEventLoop().execute(() -> {
+        boolean handled = deliverToWaiter(conn, waiter);
+        synchronized (Pool.this) {
+          waitersCount--;
+          if (!handled) {
+            synchronized (Pool.this) {
+              recycleConnection(conn, 1,false);
+              checkPending();
+            }
+          }
+        }
+      });
+      return true;
+    } else if (weight < maxWeight) {
+      weight += createConnection(waiter);
+      return true;
+    } else {
+      return false;
+    }
+  }
+
+  private void checkPending() {
+    while (waitersQueue.size() > 0) {
+      Waiter<C> waiter = waitersQueue.peek();
+      if (acquireConnection(waiter)) {
+        waitersQueue.poll();
+      } else {
+        break;
+      }
+    }
+  }
+
+  private long createConnection(Waiter<C> waiter) {
+    Holder<C> holder  = new Holder<>();
+    ConnectionListener<C> listener = new ConnectionListener<C>() {
+      @Override
+      public void onConnectSuccess(C conn, long concurrency, Channel channel, ContextImpl context, long initialWeight, long actualWeight) {
+        // Update state
+        synchronized (Pool.this) {
+          initConnection(holder, context, concurrency, conn, channel, initialWeight, actualWeight);
+        }
+        // Init connection - state might change (i.e init could close the connection)
+        waiter.initConnection(context, conn);
+        synchronized (Pool.this) {
+          if (holder.capacity == 0) {
+            waitersQueue.add(waiter);
+            checkPending();
+            return;
+          }
+          waitersCount--;
+          holder.capacity--;
+          if (holder.capacity > 0) {
+            available.add(holder);
+          }
+        }
+        boolean consumed = deliverToWaiter(holder, waiter);
+        synchronized (Pool.this) {
+          if (!consumed) {
+            recycleConnection(holder, 1,false);
+          }
+          checkPending();
+        }
+      }
+      @Override
+      public void onConnectFailure(ContextImpl context, Throwable err, long weight) {
+        waiter.handleFailure(context, err);
+        synchronized (Pool.this) {
+          waitersCount--;
+          Pool.this.weight -= weight;
+          holder.removed = true;
+          checkPending();
+          checkClose();
+        }
+      }
+      @Override
+      public void onConcurrencyChange(long concurrency) {
+        synchronized (Pool.this) {
+          if (holder.removed) {
+            return;
+          }
+          if (holder.concurrency < concurrency) {
+            long diff = concurrency - holder.concurrency;
+            if (holder.capacity == 0) {
+              available.add(holder);
+            }
+            holder.capacity += diff;
+            holder.concurrency = concurrency;
+            checkPending();
+          } else if (holder.concurrency > concurrency) {
+            throw new UnsupportedOperationException("Not yet implemented");
+          }
+        }
+      }
+      @Override
+      public void onRecycle(int capacity, boolean disposable) {
+        if (capacity < 0) {
+          throw new IllegalArgumentException("Illegal capacity");
+        }
+        synchronized (Pool.this) {
+          if (holder.removed) {
+            return;
+          }
+          recycle(holder, capacity, disposable);
+        }
+      }
+      @Override
+      public void onDiscard() {
+        synchronized (Pool.this) {
+          if (holder.removed) {
+            return;
+          }
+          closed(holder);
+        }
+      }
+    };
+    return connector.connect(listener, waiter.context);
+  }
+
+  private synchronized void recycle(Holder<C> holder, int capacity, boolean closeable) {
+    recycleConnection(holder, capacity, closeable);
+    checkPending();
+    checkClose();
+  }
+
+  private synchronized void closed(Holder<C> holder) {
+    closeConnection(holder);
+    checkPending();
+    checkClose();
+  }
+
+  private void closeConnection(Holder<C> holder) {
+    holder.removed = true;
+    connectionRemoved.accept(holder.channel, holder.connection);
+    if (holder.capacity > 0) {
+      available.remove(holder);
+      holder.capacity = 0;
+    }
+    weight -= holder.weight;
+  }
+
+  /**
+   * Should not be called under the pool lock.
+   */
+  private boolean deliverToWaiter(Holder<C> conn, Waiter<C> waiter) {
+    try {
+      return waiter.handleConnection(conn.context, conn.connection);
+    } catch (Exception e) {
+      // Handle this case gracefully
+      e.printStackTrace();
+      return true;
+    }
+  }
+
+  // These methods assume to be called under synchronization
+
+  private void recycleConnection(Holder<C> conn, int c, boolean closeable) {
+    long newCapacity = conn.capacity + c;
+    if (newCapacity > conn.concurrency) {
+      log.debug("Attempt to recycle a connection more than permitted");
+      return;
+    }
+    if (closeable && newCapacity == conn.concurrency && waitersQueue.isEmpty()) {
+      available.remove(conn);
+      conn.capacity = 0;
+      connector.close(conn.connection);
+    } else {
+      if (conn.capacity == 0) {
+        available.add(conn);
+      }
+      conn.capacity = newCapacity;
+    }
+  }
+
+  private void initConnection(Holder<C> holder, ContextImpl context, long concurrency, C conn, Channel channel, long oldWeight, long newWeight) {
+    weight += newWeight - oldWeight;
+    holder.context = context;
+    holder.concurrency = concurrency;
+    holder.connection = conn;
+    holder.channel = channel;
+    holder.weight = newWeight;
+    holder.capacity = concurrency;
+    connectionAdded.accept(holder.channel, holder.connection);
+  }
+
+  private void checkClose() {
+    if (weight == 0 && waitersCount == 0) {
+      // No waitersQueue and no connections - remove the ConnQueue
+      closed = true;
+      poolClosed.handle(null);
+    }
+  }
+}
diff --git a/src/main/java/io/vertx/core/http/impl/Waiter.java b/src/main/java/io/vertx/core/http/impl/pool/Waiter.java
similarity index 50%
rename from src/main/java/io/vertx/core/http/impl/Waiter.java
rename to src/main/java/io/vertx/core/http/impl/pool/Waiter.java
--- a/src/main/java/io/vertx/core/http/impl/Waiter.java
+++ b/src/main/java/io/vertx/core/http/impl/pool/Waiter.java
@@ -14,48 +14,45 @@
  *  You may elect to redistribute this code under either of these licenses.
  */
 
-package io.vertx.core.http.impl;
+package io.vertx.core.http.impl.pool;
 
 import io.vertx.core.impl.ContextImpl;
+import io.vertx.core.impl.ContextInternal;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-abstract class Waiter {
+public abstract class Waiter<C> {
 
-  final HttpClientRequestImpl req;
-  final ContextImpl context;
-  Object metric;
+  public final ContextImpl context;
 
-  public Waiter(HttpClientRequestImpl req, ContextImpl context) {
-    this.req = req;
+  protected Waiter(ContextImpl context) {
     this.context = context;
   }
 
   /**
-   * Handle connection failure.
+   * Handle connection failure, this callback is on a Netty even loop.
    *
+   * @param ctx the context used to create the connection
    * @param failure the failure
    */
-  abstract void handleFailure(Throwable failure);
+  public abstract void handleFailure(ContextInternal ctx, Throwable failure);
 
   /**
-   * Handle connection success.
+   * Init connection, this callback is on a Netty event loop.
    *
+   * @param ctx the context used to create the connection
    * @param conn the connection
    */
-  abstract void handleConnection(HttpClientConnection conn);
+  public abstract void initConnection(ContextInternal ctx, C conn);
 
   /**
-   * Handle connection success.
+   * Handle connection success, , this callback is on a Netty event loop.
    *
-   * @param stream the stream
+   * @param ctx the context used to create the connection
+   * @param conn the connection
+   * @return wether the waiter uses the connection
    */
-  abstract void handleStream(HttpClientStream stream);
+  public abstract boolean handleConnection(ContextInternal ctx, C conn) throws Exception;
 
-  /**
-   * @return true if the waiter has been cancelled
-   */
-  abstract boolean isCancelled();
-  
 }


Test patch:
diff --git a/src/test/benchmarks/io/vertx/benchmarks/HttpServerHandlerBenchmark.java b/src/test/benchmarks/io/vertx/benchmarks/HttpServerHandlerBenchmark.java
--- a/src/test/benchmarks/io/vertx/benchmarks/HttpServerHandlerBenchmark.java
+++ b/src/test/benchmarks/io/vertx/benchmarks/HttpServerHandlerBenchmark.java
@@ -19,7 +19,6 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
-import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.embedded.EmbeddedChannel;
@@ -42,7 +41,7 @@
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.HttpServerResponse;
 import io.vertx.core.http.impl.HttpHandlers;
-import io.vertx.core.http.impl.ServerHandler;
+import io.vertx.core.http.impl.Http1xServerHandler;
 import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.impl.EventLoopContext;
 import io.vertx.core.impl.VertxInternal;
@@ -232,7 +231,7 @@ public void setup() {
       response.end(HELLO_WORLD_BUFFER);
     };
     HandlerHolder<HttpHandlers> holder = new HandlerHolder<>(context, new HttpHandlers(app, null, null, null));
-    ServerHandler handler = new ServerHandler(null, new HttpServerOptions(), "localhost", holder, null);
+    Http1xServerHandler handler = new Http1xServerHandler(null, new HttpServerOptions(), "localhost", holder, null);
     vertxChannel.pipeline().addLast("handler", handler);
 
     nettyChannel = new EmbeddedChannel(new HttpRequestDecoder(
diff --git a/src/test/java/io/vertx/test/ListTransferTest.java b/src/test/java/io/vertx/test/ListTransferTest.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/test/ListTransferTest.java
@@ -0,0 +1,81 @@
+package io.vertx.test;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+
+public class ListTransferTest {
+
+  private static final int CONNECTING = 0;
+  private static final int WRITING = 1;
+  private static final int CONNECTED = 2;
+
+  @Test
+  public void doTheTest() throws Exception {
+
+    AtomicInteger status = new AtomicInteger();
+    ArrayList<Integer> array = new ArrayList<>();
+
+    AtomicInteger expectedCount = new AtomicInteger();
+
+    Thread t1 = new Thread(() -> {
+      int count = 0;
+      while (true) {
+        switch (status.get()) {
+          case CONNECTING:
+            if (status.compareAndSet(CONNECTING, WRITING)) {
+              array.add(count++);
+              status.set(CONNECTING);
+            }
+            break;
+          case CONNECTED: {
+            expectedCount.set(count);
+            return;
+          }
+        }
+      }
+    });
+
+    AtomicReference<ArrayList<Integer>> expectedArray = new AtomicReference<>();
+    AtomicInteger expectedSpin = new AtomicInteger();
+    Thread t2 = new Thread(() -> {
+      try {
+        Thread.sleep(50);
+      } catch (InterruptedException ignore) {
+        return;
+      }
+      int spin = 0;
+      while (true) {
+        if (status.compareAndSet(CONNECTING, CONNECTED)) {
+          expectedArray.set(new ArrayList<>(array));
+          expectedSpin.set(spin);
+          return;
+        } else {
+          spin ++;
+          Thread.yield();
+        }
+      }
+    });
+
+    t1.start();
+    t2.start();
+
+    t1.join();
+    t2.join();
+
+    assertEquals(expectedCount.get(), expectedArray.get().size());
+
+    System.out.println(expectedCount.get());
+    System.out.println(expectedSpin.get());
+
+
+
+
+  }
+
+
+}
diff --git a/src/test/java/io/vertx/test/core/Http1xTest.java b/src/test/java/io/vertx/test/core/Http1xTest.java
--- a/src/test/java/io/vertx/test/core/Http1xTest.java
+++ b/src/test/java/io/vertx/test/core/Http1xTest.java
@@ -25,20 +25,7 @@
 import io.vertx.core.Vertx;
 import io.vertx.core.VertxException;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.ConnectionPoolTooBusyException;
-import io.vertx.core.http.Http2Settings;
-import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientOptions;
-import io.vertx.core.http.HttpClientRequest;
-import io.vertx.core.http.HttpConnection;
-import io.vertx.core.http.HttpHeaders;
-import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.HttpServer;
-import io.vertx.core.http.HttpServerOptions;
-import io.vertx.core.http.HttpServerRequest;
-import io.vertx.core.http.HttpServerResponse;
-import io.vertx.core.http.HttpVersion;
-import io.vertx.core.http.RequestOptions;
+import io.vertx.core.http.*;
 import io.vertx.core.http.impl.HttpClientRequestImpl;
 import io.vertx.core.impl.ConcurrentHashSet;
 import io.vertx.core.impl.ContextImpl;
@@ -1250,6 +1237,47 @@ public void testPipeliningLimit() throws Exception {
     await();
   }
 
+  @Test
+  public void testPipeliningFailure() throws Exception {
+    int n = 5;
+    client.close();
+    client = vertx.createHttpClient(new HttpClientOptions().setKeepAlive(true).setPipelining(true).setPipeliningLimit(n).setMaxPoolSize(1));
+    CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+    AtomicBoolean first = new AtomicBoolean(true);
+    server.requestHandler(req -> {
+      if (first.compareAndSet(true, false)) {
+        closeFuture.thenAccept(v -> {
+          req.response().close();
+        });
+      } else {
+        req.response().end();
+      }
+    });
+    startServer();
+    AtomicInteger succeeded = new AtomicInteger();
+    List<HttpClientRequest> requests = new ArrayList<>();
+    for (int i = 0;i < n * 2;i++) {
+      HttpClientRequest req = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/" + i);
+      req.handler(resp -> {
+        succeeded.incrementAndGet();
+        requests.remove(req);
+        if (requests.isEmpty()) {
+          assertEquals(n * 2 - 1, succeeded.get());
+          testComplete();
+        }
+      });
+      req.exceptionHandler(err -> {
+        requests.remove(req);
+        for (HttpClientRequest r : requests) {
+          r.end();
+        }
+      }).sendHead();
+      requests.add(req);
+    }
+    closeFuture.complete(null);
+    await();
+  }
+
   @Test
   public void testKeepAlive() throws Exception {
     testKeepAlive(true, 5, 10, 5);
@@ -1334,11 +1362,6 @@ public void testPoolingNoKeepAliveNoPipelining() {
     testPooling(false, false);
   }
 
-  @Test
-  public void testPoolingNoKeepAliveAndPipelining() {
-    testPooling(false, true);
-  }
-
   private void testPooling(boolean keepAlive, boolean pipelining) {
     String path = "foo.txt";
     int numGets = 100;
@@ -1367,15 +1390,7 @@ private void testPooling(boolean keepAlive, boolean pipelining) {
           }
         });
         req.exceptionHandler(t -> {
-          if (pipelining && !keepAlive) {
-            // Illegal combination - should get exception
-            assertTrue(t instanceof IllegalStateException);
-            if (completeAlready.compareAndSet(false, true)) {
-              testComplete();
-            }
-          } else {
-            fail("Should not throw exception: " + t.getMessage());
-          }
+          fail("Should not throw exception: " + t.getMessage());
         });
         req.headers().set("count", String.valueOf(i));
         req.end();
@@ -1385,6 +1400,15 @@ private void testPooling(boolean keepAlive, boolean pipelining) {
     await();
   }
 
+  @Test
+  public void testPoolingNoKeepAliveAndPipelining() {
+    try {
+      vertx.createHttpClient(new HttpClientOptions().setKeepAlive(false).setPipelining(true));
+      fail();
+    } catch (IllegalStateException ignore) {
+    }
+  }
+
   @Test
   public void testMaxWaitQueueSizeIsRespected() throws Exception {
     client.close();
@@ -1392,38 +1416,35 @@ public void testMaxWaitQueueSizeIsRespected() throws Exception {
     client = vertx.createHttpClient(new HttpClientOptions().setDefaultHost(DEFAULT_HTTP_HOST).setDefaultPort(DEFAULT_HTTP_PORT)
         .setPipelining(false).setMaxWaitQueueSize(0).setMaxPoolSize(2));
 
+    waitFor(3);
+
+    Set<String> expected = new HashSet<>(Arrays.asList("/1", "/2"));
     server.requestHandler(req -> {
-      req.response().setStatusCode(200);
-      req.response().end("OK");
+      assertTrue(expected.contains(req.path()));
+      complete();
     });
 
-    server.listen(onSuccess(s -> {
-      HttpClientRequest req1 = client.get(DEFAULT_TEST_URI, resp -> {
-        resp.bodyHandler(body -> {
-          assertEquals("OK", body.toString());
-        });
-      });
-      req1.exceptionHandler(t -> fail("Should not be called."));
+    startServer();
 
-      HttpClientRequest req2 = client.get(DEFAULT_TEST_URI, resp -> {
-        resp.bodyHandler(body -> {
-          assertEquals("OK", body.toString());
-          testComplete();
-        });
-      });
-      req2.exceptionHandler(t -> fail("Should not be called."));
+    HttpClientRequest req1 = client.get("/1", resp -> {
+      fail("Should not be called.");
+    });
 
-      HttpClientRequest req3 = client.get(DEFAULT_TEST_URI, resp -> {
-        fail("Should not be called.");
-      });
-      req3.exceptionHandler(t -> {
-        assertTrue("Incorrect exception time.", t instanceof ConnectionPoolTooBusyException);
-      });
+    HttpClientRequest req2 = client.get("/2", resp -> {
+      fail("Should not be called.");
+    });
 
-      req1.end();
-      req2.end();
-      req3.end();
-    }));
+    HttpClientRequest req3 = client.get("/3", resp -> {
+      fail("Should not be called.");
+    });
+    req3.exceptionHandler(t -> {
+      assertTrue("Incorrect exception: " + t.getClass().getName(), t instanceof ConnectionPoolTooBusyException);
+      complete();
+    });
+
+    req1.end();
+    req2.end();
+    req3.end();
 
     await();
   }
@@ -1614,7 +1635,7 @@ public void testDefaultHttpVersion() {
   public void testIncorrectHttpVersion() throws Exception {
     server.requestHandler(req -> {
       NetSocket so = req.netSocket();
-      so.write(Buffer.buffer("HTTP/1.2 200 OK\r\n\r\n"));
+      so.write(Buffer.buffer("HTTP/1.2 200 OK\r\nContent-Length:5\r\n\r\nHELLO"));
       so.close();
     });
     startServer();
@@ -1891,60 +1912,40 @@ public void testClientContextWithPipelining() throws Exception {
   }
 
   private void testClientContext() throws Exception {
-    CountDownLatch serverLatch = new CountDownLatch(1);
-    CountDownLatch req1Latch = new CountDownLatch(1);
     server.requestHandler(req -> {
       req.response().end();
-      req1Latch.countDown();
-    } ).listen(ar -> {
-      assertTrue(ar.succeeded());
-      serverLatch.countDown();
-    });
-    awaitLatch(serverLatch);
-    AtomicReference<Context> c = new AtomicReference<>();
-    HttpClientRequest req1 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/1");
-    AtomicReference<HttpConnection> conn = new AtomicReference<>();
-    req1.handler(res -> {
-      c.set(Vertx.currentContext());
-      conn.set(req1.connection());
     });
-    req1.end();
-    Consumer<HttpClientRequest> checker = req -> {
-      assertSame(Vertx.currentContext(), c.get());
-      assertSame(conn.get(), req.connection());
+    startServer();
+    Set<Context> contexts = Collections.synchronizedSet(new HashSet<>());
+    Set<HttpConnection> connections = Collections.synchronizedSet(new HashSet<>());
+    Handler<HttpClientResponse> checker = response -> {
+      contexts.add(Vertx.currentContext());
+      connections.add(response.request().connection());
     };
-    CountDownLatch req2Latch = new CountDownLatch(2);
-    HttpClientRequest req2 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/2");
-    req2.handler(res -> {
-      checker.accept(req2);
-      req2Latch.countDown();
-    }).exceptionHandler(err -> {
-      fail(err);
-    }).sendHead();
-    awaitLatch(req1Latch);
-    HttpClientRequest req3 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/3");
-    req3.handler(res -> {
-      checker.accept(req3);
-      req2Latch.countDown();
-    }).exceptionHandler(err -> {
-      fail(err);
-    });
-    req2.end();
-    req3.end();
-    awaitLatch(req2Latch);
-    vertx.getOrCreateContext().runOnContext(v -> {
-      HttpClientRequest req4 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/4");
-      req4.handler(res -> {
+    HttpClientRequest req1 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/2");
+    req1.handler(checker).exceptionHandler(this::fail);
+    HttpClientRequest req2 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/3");
+    req2.handler(checker).exceptionHandler(this::fail);
+    CompletableFuture<HttpClientRequest> fut = new CompletableFuture<>();
+    Context ctx = vertx.getOrCreateContext();
+    ctx.runOnContext(v -> {
+      HttpClientRequest req3 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/4");
+      req3.handler(resp -> {
         // This should warn in the log (console) as we are called back on the connection context
         // and not on the context doing the request
-        checker.accept(req4);
+        // checker.accept(req4);
+        assertEquals(1, contexts.size());
+        assertEquals(1, connections.size());
+        assertNotSame(Vertx.currentContext(), ctx);
         testComplete();
       });
-      req4.exceptionHandler(err -> {
-        fail(err);
-      });
-      req4.end();
+      req3.exceptionHandler(this::fail);
+      fut.complete(req3);
     });
+    HttpClientRequest req3 = fut.get(10, TimeUnit.SECONDS);
+    req1.end();
+    req2.end();
+    req3.end();
     await();
   }
 
@@ -2530,11 +2531,11 @@ public void testConnectionCloseHttp_1_1_NoClose() throws Exception {
       AtomicBoolean firstRequest = new AtomicBoolean(true);
       socket.handler(RecordParser.newDelimited("\r\n\r\n", buffer -> {
         if (firstRequest.getAndSet(false)) {
-          socket.write("HTTP/1.1 200 OK\n" + "Content-Type: text/plain\n" + "Content-Length: 4\n"
-              + "\n" + "xxx\n");
+          socket.write("HTTP/1.1 200 OK\r\n" + "Content-Type: text/plain\r\n" + "Content-Length: 4\r\n"
+              + "\r\n" + "xxx\n");
         } else {
-          socket.write("HTTP/1.1 200 OK\n" + "Content-Type: text/plain\n" + "Content-Length: 1\n"
-              + "Connection: close\n" + "\n" + "\n");
+          socket.write("HTTP/1.1 200 OK\r\n" + "Content-Type: text/plain\r\n" + "Content-Length: 1\r\n"
+              + "Connection: close\r\n" + "\r\n" + "\r\n");
         }
       }));
     });
@@ -2546,11 +2547,11 @@ public void testConnectionCloseHttp_1_1_Close() throws Exception {
       AtomicBoolean firstRequest = new AtomicBoolean(true);
       socket.handler(RecordParser.newDelimited("\r\n\r\n", buffer -> {
         if (firstRequest.getAndSet(false)) {
-          socket.write("HTTP/1.1 200 OK\n" + "Content-Type: text/plain\n" + "Content-Length: 4\n"
-              + "\n" + "xxx\n");
+          socket.write("HTTP/1.1 200 OK\r\n" + "Content-Type: text/plain\r\n" + "Content-Length: 3\r\n"
+              + "\r\n" + "xxx");
         } else {
-          socket.write("HTTP/1.1 200 OK\n" + "Content-Type: text/plain\n" + "Content-Length: 1\n"
-              + "Connection: close\n" + "\n" + "\n");
+          socket.write("HTTP/1.1 200 OK\r\n" + "Content-Type: text/plain\r\n" + "Content-Length: 0\r\n"
+              + "Connection: close\r\n" + "\r\n");
           socket.close();
         }
       }));
@@ -3016,7 +3017,7 @@ public void testResetKeepAliveClientRequest() throws Exception {
   @Test
   public void testResetPipelinedClientRequest() throws Exception {
     waitFor(2);
-    CompletableFuture<Void> reset = new CompletableFuture<>();
+    CompletableFuture<Void> doReset = new CompletableFuture<>();
     server.close();
     NetServer server = vertx.createNetServer();
     AtomicInteger count = new AtomicInteger();
@@ -3031,7 +3032,7 @@ public void testResetPipelinedClientRequest() throws Exception {
             "POST /somepath HTTP/1.1\r\n" +
             "Host: localhost:8080\r\n" +
             "\r\n")) {
-          reset.complete(null);
+          doReset.complete(null);
           so.write(
               "HTTP/1.1 200 OK\r\n" +
               "Content-Type: text/plain\r\n" +
@@ -3052,13 +3053,11 @@ public void testResetPipelinedClientRequest() throws Exception {
       awaitLatch(listenLatch);
       client.close();
       client = vertx.createHttpClient(new HttpClientOptions().setMaxPoolSize(1).setPipelining(true).setKeepAlive(true));
-      AtomicInteger status = new AtomicInteger();
       HttpClientRequest req1 = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/somepath", resp -> {
-        assertEquals(0, status.getAndIncrement());
+        fail();
       });
       req1.connectionHandler(conn -> {
         conn.closeHandler(v -> {
-          assertEquals(1, status.getAndIncrement());
           complete();
         });
       });
@@ -3067,7 +3066,7 @@ public void testResetPipelinedClientRequest() throws Exception {
         fail();
       });
       req2.sendHead();
-      reset.thenAccept(v -> {
+      doReset.thenAccept(v -> {
         assertTrue(req2.reset());
       });
       await();
diff --git a/src/test/java/io/vertx/test/core/Http2ClientTest.java b/src/test/java/io/vertx/test/core/Http2ClientTest.java
--- a/src/test/java/io/vertx/test/core/Http2ClientTest.java
+++ b/src/test/java/io/vertx/test/core/Http2ClientTest.java
@@ -71,6 +71,7 @@
 import io.vertx.core.net.impl.SSLHelper;
 import io.vertx.test.core.tls.Cert;
 import io.vertx.test.core.tls.Trust;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import java.io.ByteArrayOutputStream;
@@ -134,20 +135,19 @@ public void testClientSettings() throws Exception {
         req.response().end();
       });
     }).connectionHandler(conn -> {
+      io.vertx.core.http.Http2Settings initialRemoteSettings = conn.remoteSettings();
+      assertEquals(initialSettings.isPushEnabled(), initialRemoteSettings.isPushEnabled());
+      assertEquals(initialSettings.getMaxHeaderListSize(), initialRemoteSettings.getMaxHeaderListSize());
+      assertEquals(initialSettings.getMaxFrameSize(), initialRemoteSettings.getMaxFrameSize());
+      assertEquals(initialSettings.getInitialWindowSize(), initialRemoteSettings.getInitialWindowSize());
+//            assertEquals(Math.min(initialSettings.getMaxConcurrentStreams(), Integer.MAX_VALUE), settings.getMaxConcurrentStreams());
+      assertEquals(initialSettings.getHeaderTableSize(), initialRemoteSettings.getHeaderTableSize());
+      assertEquals(initialSettings.get('\u0007'), initialRemoteSettings.get(7));
       Context ctx = Vertx.currentContext();
       conn.remoteSettingsHandler(settings -> {
         assertOnIOContext(ctx);
         switch (count.getAndIncrement()) {
           case 0:
-            assertEquals(initialSettings.isPushEnabled(), settings.isPushEnabled());
-            assertEquals(initialSettings.getMaxHeaderListSize(), settings.getMaxHeaderListSize());
-            assertEquals(initialSettings.getMaxFrameSize(), settings.getMaxFrameSize());
-            assertEquals(initialSettings.getInitialWindowSize(), settings.getInitialWindowSize());
-//            assertEquals(Math.min(initialSettings.getMaxConcurrentStreams(), Integer.MAX_VALUE), settings.getMaxConcurrentStreams());
-            assertEquals(initialSettings.getHeaderTableSize(), settings.getHeaderTableSize());
-            assertEquals(initialSettings.get('\u0007'), settings.get(7));
-            break;
-          case 1:
             // find out why it fails sometimes ...
             // assertEquals(updatedSettings.pushEnabled(), settings.getEnablePush());
             assertEquals(updatedSettings.getMaxHeaderListSize(), settings.getMaxHeaderListSize());
@@ -159,6 +159,9 @@ public void testClientSettings() throws Exception {
             assertEquals(updatedSettings.get('\u0007'), settings.get(7));
             complete();
             break;
+          default:
+            fail();
+
         }
       });
     });
@@ -198,7 +201,6 @@ public void testInvalidSettings() throws Exception {
 
   @Test
   public void testServerSettings() throws Exception {
-    waitFor(2);
     io.vertx.core.http.Http2Settings expectedSettings = TestUtils.randomHttp2Settings();
     expectedSettings.setHeaderTableSize((int)io.vertx.core.http.Http2Settings.DEFAULT_HEADER_TABLE_SIZE);
     server.close();
@@ -210,26 +212,20 @@ public void testServerSettings() throws Exception {
       });
     });
     server.requestHandler(req -> {
-      req.response().end();
     });
     startServer();
     AtomicInteger count = new AtomicInteger();
-    client.get(DEFAULT_HTTPS_PORT, DEFAULT_HTTPS_HOST, "/somepath", resp -> {
-      complete();
-    }).connectionHandler(conn -> {
+    client.get(DEFAULT_HTTPS_PORT, DEFAULT_HTTPS_HOST, "/somepath", resp -> fail()).connectionHandler(conn -> {
       conn.remoteSettingsHandler(settings -> {
         switch (count.getAndIncrement()) {
           case 0:
-            // Initial settings
-            break;
-          case 1:
             assertEquals(expectedSettings.getMaxHeaderListSize(), settings.getMaxHeaderListSize());
             assertEquals(expectedSettings.getMaxFrameSize(), settings.getMaxFrameSize());
             assertEquals(expectedSettings.getInitialWindowSize(), settings.getInitialWindowSize());
             assertEquals(expectedSettings.getMaxConcurrentStreams(), settings.getMaxConcurrentStreams());
             assertEquals(expectedSettings.getHeaderTableSize(), settings.getHeaderTableSize());
             assertEquals(expectedSettings.get('\u0007'), settings.get(7));
-            complete();
+            testComplete();
             break;
         }
       });
@@ -594,10 +590,8 @@ private void testQueueingRequests(int numReq, Long max) throws Exception {
     CountDownLatch latch = new CountDownLatch(1);
     client.get(DEFAULT_HTTPS_PORT, DEFAULT_HTTPS_HOST, "/somepath", resp -> {
     }).connectionHandler(conn -> {
-      conn.remoteSettingsHandler(settings -> {
-        assertEquals(max == null ? 0xFFFFFFFFL : max, settings.getMaxConcurrentStreams());
-        latch.countDown();
-      });
+      assertEquals(max == null ? 0xFFFFFFFFL : max, conn.remoteSettings().getMaxConcurrentStreams());
+      latch.countDown();
     }).exceptionHandler(err -> {
       fail();
     }).end();
@@ -946,6 +940,7 @@ public void testConnectionHandler() throws Exception {
 
   @Test
   public void testConnectionShutdownInConnectionHandler() throws Exception {
+    waitFor(2);
     AtomicInteger serverStatus = new AtomicInteger();
     server.connectionHandler(conn -> {
       if (serverStatus.getAndIncrement() == 0) {
@@ -963,28 +958,27 @@ public void testConnectionShutdownInConnectionHandler() throws Exception {
     });
     server.requestHandler(req -> {
       assertEquals(5, serverStatus.getAndIncrement());
-      req.response().end();
+      req.response().end("" + serverStatus.get());
     });
     startServer();
     AtomicInteger clientStatus = new AtomicInteger();
     HttpClientRequest req1 = client.get(DEFAULT_HTTPS_PORT, DEFAULT_HTTPS_HOST, "/somepath");
     req1.connectionHandler(conn -> {
       Context ctx = Vertx.currentContext();
-      conn.shutdownHandler(v -> {
-        assertOnIOContext(ctx);
-        clientStatus.compareAndSet(1, 2);
-      });
       if (clientStatus.getAndIncrement() == 0) {
+        conn.shutdownHandler(v -> {
+          assertOnIOContext(ctx);
+          clientStatus.compareAndSet(1, 2);
+          complete();
+        });
         conn.shutdown();
       }
     });
-    req1.exceptionHandler(err -> {
-      fail();
-    });
+    req1.exceptionHandler(err -> fail());
     req1.handler(resp -> {
-      assertEquals(2, clientStatus.getAndIncrement());
-      resp.endHandler(v -> {
-        testComplete();
+      resp.bodyHandler(body -> {
+        assertEquals("6", body.toString());
+        complete();
       });
     });
     req1.end();
@@ -1021,7 +1015,9 @@ public void testServerShutdownConnection() throws Exception {
   @Test
   public void testReceivingGoAwayDiscardsTheConnection() throws Exception {
     AtomicInteger reqCount = new AtomicInteger();
+    Set<HttpConnection> connections = Collections.synchronizedSet(new HashSet<>());
     server.requestHandler(req -> {
+      connections.add(req.connection());
       switch (reqCount.getAndIncrement()) {
         case 0:
           req.connection().goAway(0);
@@ -1040,6 +1036,7 @@ public void testReceivingGoAwayDiscardsTheConnection() throws Exception {
       conn.goAwayHandler(ga -> {
         if (gaCount.getAndIncrement() == 0) {
           client.get(DEFAULT_HTTPS_PORT, DEFAULT_HTTPS_HOST, "/somepath", resp2 -> {
+            assertEquals(2, connections.size());
             testComplete();
           }).setTimeout(5000).exceptionHandler(this::fail).end();
         }
@@ -1748,14 +1745,13 @@ public void testMaxConcurrencySingleConnection() throws Exception {
 
   @Test
   public void testMaxConcurrencyMultipleConnections() throws Exception {
-    testMaxConcurrency(3, 5);
+    testMaxConcurrency(2, 1);
   }
 
   private void testMaxConcurrency(int poolSize, int maxConcurrency) throws Exception {
     int rounds = 1 + poolSize;
     int maxRequests = poolSize * maxConcurrency;
     int totalRequests = maxRequests + maxConcurrency;
-
     Set<HttpConnection> serverConns = new HashSet<>();
     server.connectionHandler(conn -> {
       serverConns.add(conn);
@@ -1781,7 +1777,6 @@ private void testMaxConcurrency(int poolSize, int maxConcurrency) throws Excepti
         setHttp2MaxPoolSize(poolSize).
         setHttp2MultiplexingLimit(maxConcurrency));
     AtomicInteger respCount = new AtomicInteger();
-
     Set<HttpConnection> clientConnections = Collections.synchronizedSet(new HashSet<>());
     for (int j = 0;j < rounds;j++) {
       for (int i = 0;i < maxConcurrency;i++) {
diff --git a/src/test/java/io/vertx/test/core/Http2ServerTest.java b/src/test/java/io/vertx/test/core/Http2ServerTest.java
--- a/src/test/java/io/vertx/test/core/Http2ServerTest.java
+++ b/src/test/java/io/vertx/test/core/Http2ServerTest.java
@@ -386,34 +386,24 @@ public void testClientSettings() throws Exception {
     Context ctx = vertx.getOrCreateContext();
     io.vertx.core.http.Http2Settings initialSettings = TestUtils.randomHttp2Settings();
     io.vertx.core.http.Http2Settings updatedSettings = TestUtils.randomHttp2Settings();
-    Future<Void> settingsRead = Future.future();
     AtomicInteger count = new AtomicInteger();
     server.connectionHandler(conn -> {
       io.vertx.core.http.Http2Settings settings = conn.remoteSettings();
-      assertEquals(true, settings.isPushEnabled());
+      assertEquals(initialSettings.isPushEnabled(), settings.isPushEnabled());
 
       // Netty bug ?
       // Nothing has been yet received so we should get Integer.MAX_VALUE
       // assertEquals(Integer.MAX_VALUE, settings.getMaxHeaderListSize());
 
-      assertEquals(io.vertx.core.http.Http2Settings.DEFAULT_MAX_FRAME_SIZE, settings.getMaxFrameSize());
-      assertEquals(io.vertx.core.http.Http2Settings.DEFAULT_INITIAL_WINDOW_SIZE, settings.getInitialWindowSize());
-      assertEquals((Long)(long)Integer.MAX_VALUE, (Long)(long)settings.getMaxConcurrentStreams());
-      assertEquals(io.vertx.core.http.Http2Settings.DEFAULT_HEADER_TABLE_SIZE, settings.getHeaderTableSize());
+      assertEquals(initialSettings.getMaxFrameSize(), settings.getMaxFrameSize());
+      assertEquals(initialSettings.getInitialWindowSize(), settings.getInitialWindowSize());
+      assertEquals((Long)(long)initialSettings.getMaxConcurrentStreams(), (Long)(long)settings.getMaxConcurrentStreams());
+      assertEquals(initialSettings.getHeaderTableSize(), settings.getHeaderTableSize());
+
       conn.remoteSettingsHandler(update -> {
         assertOnIOContext(ctx);
         switch (count.getAndIncrement()) {
           case 0:
-            assertEquals(initialSettings.isPushEnabled(), update.isPushEnabled());
-            assertEquals(initialSettings.getMaxHeaderListSize(), update.getMaxHeaderListSize());
-            assertEquals(initialSettings.getMaxFrameSize(), update.getMaxFrameSize());
-            assertEquals(initialSettings.getInitialWindowSize(), update.getInitialWindowSize());
-            assertEquals(initialSettings.getMaxConcurrentStreams(), update.getMaxConcurrentStreams());
-            assertEquals(initialSettings.getHeaderTableSize(), update.getHeaderTableSize());
-            assertEquals(initialSettings.get('\u0007'), update.get(7));
-            settingsRead.complete();
-            break;
-          case 1:
             assertEquals(updatedSettings.isPushEnabled(), update.isPushEnabled());
             assertEquals(updatedSettings.getMaxHeaderListSize(), update.getMaxHeaderListSize());
             assertEquals(updatedSettings.getMaxFrameSize(), update.getMaxFrameSize());
@@ -423,6 +413,8 @@ public void testClientSettings() throws Exception {
             assertEquals(updatedSettings.get('\u0007'), update.get(7));
             testComplete();
             break;
+          default:
+            fail();
         }
       });
     });
@@ -433,10 +425,8 @@ public void testClientSettings() throws Exception {
     TestClient client = new TestClient();
     client.settings.putAll(HttpUtils.fromVertxSettings(initialSettings));
     ChannelFuture fut = client.connect(DEFAULT_HTTPS_PORT, DEFAULT_HTTPS_HOST, request -> {
-      settingsRead.setHandler(ar -> {
-        request.encoder.writeSettings(request.context, HttpUtils.fromVertxSettings(updatedSettings), request.context.newPromise());
-        request.context.flush();
-      });
+      request.encoder.writeSettings(request.context, HttpUtils.fromVertxSettings(updatedSettings), request.context.newPromise());
+      request.context.flush();
     });
     fut.sync();
     await();
diff --git a/src/test/java/io/vertx/test/core/Http2Test.java b/src/test/java/io/vertx/test/core/Http2Test.java
--- a/src/test/java/io/vertx/test/core/Http2Test.java
+++ b/src/test/java/io/vertx/test/core/Http2Test.java
@@ -25,11 +25,13 @@
 import io.vertx.test.core.tls.Cert;
 import org.junit.Test;
 
+import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
@@ -356,4 +358,51 @@ public void testClientMakeRequestHttp2WithSSLWithoutAlpn() throws Exception {
       // Expected
     }
   }
+
+  @Test
+  public void testServePendingRequests() throws Exception {
+    int n = 10;
+    waitFor(n);
+    LinkedList<HttpServerRequest> requests = new LinkedList<>();
+    Set<HttpConnection> connections = new HashSet<>();
+    server.requestHandler(req -> {
+      requests.add(req);
+      connections.add(req.connection());
+      assertEquals(1, connections.size());
+      if (requests.size() == n) {
+        while (requests.size() > 0) {
+          requests.removeFirst().response().end();
+        }
+      }
+    });
+    startServer();
+    for (int i = 0;i < n;i++) {
+      client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> complete()).end();
+    }
+    await();
+  }
+
+  @Test
+  public void testInitialMaxConcurrentStreamZero() throws Exception {
+    AtomicLong concurrency = new AtomicLong();
+    server.close();
+    server = vertx.createHttpServer(createBaseServerOptions().setInitialSettings(new Http2Settings().setMaxConcurrentStreams(0)));
+    server.requestHandler(req -> {
+      assertEquals(10, concurrency.get());
+      req.response().end();
+    });
+    server.connectionHandler(conn -> {
+      vertx.setTimer(500, id -> {
+        conn.updateSettings(new Http2Settings().setMaxConcurrentStreams(10));
+      });
+    });
+    startServer();
+    client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> {
+      testComplete();
+    }).connectionHandler(conn -> {
+      assertEquals(0, conn.remoteSettings().getMaxConcurrentStreams());
+      conn.remoteSettingsHandler(settings -> concurrency.set(settings.getMaxConcurrentStreams()));
+    }).setTimeout(10000).exceptionHandler(err -> fail(err)).end();
+    await();
+  }
 }
diff --git a/src/test/java/io/vertx/test/core/HttpTLSTest.java b/src/test/java/io/vertx/test/core/HttpTLSTest.java
--- a/src/test/java/io/vertx/test/core/HttpTLSTest.java
+++ b/src/test/java/io/vertx/test/core/HttpTLSTest.java
@@ -1216,10 +1216,8 @@ public void testCrlInvalidPath() throws Exception {
     clientOptions.setTrustOptions(Trust.SERVER_PEM_ROOT_CA.get());
     clientOptions.setSsl(true);
     clientOptions.addCrlPath("/invalid.pem");
-    HttpClient client = vertx.createHttpClient(clientOptions);
-    HttpClientRequest req = client.request(HttpMethod.CONNECT, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/", (handler) -> {});
     try {
-      req.end();
+      vertx.createHttpClient(clientOptions);
       fail("Was expecting a failure");
     } catch (VertxException e) {
       assertNotNull(e.getCause());
diff --git a/src/test/java/io/vertx/test/core/HttpTest.java b/src/test/java/io/vertx/test/core/HttpTest.java
--- a/src/test/java/io/vertx/test/core/HttpTest.java
+++ b/src/test/java/io/vertx/test/core/HttpTest.java
@@ -16,8 +16,10 @@
 
 package io.vertx.test.core;
 
+import io.netty.handler.codec.compression.DecompressionException;
 import io.netty.handler.codec.http.DefaultHttpHeaders;
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http2.Http2Exception;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 import io.vertx.codegen.annotations.Nullable;
 import io.vertx.core.AbstractVerticle;
@@ -47,30 +49,17 @@
 import io.vertx.core.impl.WorkerContext;
 import io.vertx.core.net.NetClient;
 import io.vertx.core.net.NetSocket;
+import io.vertx.core.net.SocketAddress;
 import io.vertx.test.netty.TestLoggerFactory;
 import org.junit.Assume;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
-import java.io.OutputStreamWriter;
-import java.io.UnsupportedEncodingException;
-import java.net.InetAddress;
-import java.net.URLEncoder;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.io.*;
+import java.net.*;
+import java.util.*;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -2400,7 +2389,29 @@ public void testPauseClientResponse() {
 
   @Test
   public void testDeliverPausedBufferWhenResume() throws Exception {
-    Buffer data = TestUtils.randomBuffer(20);
+    testDeliverPausedBufferWhenResume(block -> vertx.setTimer(10, id -> block.run()));
+  }
+
+  @Test
+  public void testDeliverPausedBufferWhenResumeOnOtherThread() throws Exception {
+    ExecutorService exec = Executors.newSingleThreadExecutor();
+    try {
+      testDeliverPausedBufferWhenResume(block -> exec.execute(() -> {
+        try {
+          Thread.sleep(10);
+        } catch (InterruptedException e) {
+          fail(e);
+          Thread.currentThread().interrupt();
+        }
+        block.run();
+      }));
+    } finally {
+      exec.shutdown();
+    }
+  }
+
+  private void testDeliverPausedBufferWhenResume(Consumer<Runnable> scheduler) throws Exception {
+    Buffer data = TestUtils.randomBuffer(2048);
     int num = 10;
     waitFor(num);
     List<CompletableFuture<Void>> resumes = Collections.synchronizedList(new ArrayList<>());
@@ -2422,18 +2433,18 @@ public void testDeliverPausedBufferWhenResume() throws Exception {
       int idx = i;
       client.request(HttpMethod.GET, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/" + i, resp -> {
         Buffer body = Buffer.buffer();
+        Thread t = Thread.currentThread();
         resp.handler(buff -> {
+          assertSame(t, Thread.currentThread());
           resumes.get(idx).complete(null);
           body.appendBuffer(buff);
         });
         resp.endHandler(v -> {
-          assertEquals(data, body);
+          // assertEquals(data, body);
           complete();
         });
         resp.pause();
-        vertx.setTimer(10, id -> {
-          resp.resume();
-        });
+        scheduler.accept(resp::resume);
       }).end();
     }
     await();
@@ -3858,6 +3869,37 @@ private TestLoggerFactory testLogging() throws Exception {
     return factory;
   }
 
+  @Test
+  public void testClientDecompressionError() throws Exception {
+    waitFor(2);
+    server.requestHandler(req -> {
+      req.response()
+        .putHeader("Content-Encoding", "gzip")
+        .end("long response with mismatched encoding causes connection leaks");
+    });
+    startServer();
+    client.close();
+    client = vertx.createHttpClient(createBaseClientOptions().setTryUseCompression(true));
+    client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> {
+      resp.exceptionHandler(err -> {
+        if (err instanceof Http2Exception) {
+          complete();
+          // Connection is not closed for HTTP/2 only the streams so we need to force it
+          resp.request().connection().close();
+        } else if (err instanceof DecompressionException) {
+          complete();
+        }
+      });
+    }).connectionHandler(conn -> {
+      conn.closeHandler(v -> {
+        complete();
+      });
+    }).end();
+
+    await();
+
+  }
+
   protected File setupFile(String fileName, String content) throws Exception {
     File file = new File(testDir, fileName);
     if (file.exists()) {
@@ -3901,4 +3943,26 @@ protected static MultiMap getHeaders(int num) {
     }
     return headers;
   }
+/*
+  @Test
+  public void testReset() throws Exception {
+    CountDownLatch latch = new CountDownLatch(1);
+    server.requestHandler(req -> {
+      req.exceptionHandler(err -> {
+        System.out.println("GOT ERR");
+      });
+      req.endHandler(v -> {
+        System.out.println("GOT END");
+        latch.countDown();
+      });
+    });
+    startServer();
+    HttpClientRequest req = client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/somepath", resp -> {});
+    req.end();
+    awaitLatch(latch);
+    req.reset();
+
+    await();
+  }
+*/
 }
diff --git a/src/test/java/io/vertx/test/core/MetricsTest.java b/src/test/java/io/vertx/test/core/MetricsTest.java
--- a/src/test/java/io/vertx/test/core/MetricsTest.java
+++ b/src/test/java/io/vertx/test/core/MetricsTest.java
@@ -680,7 +680,7 @@ public void testHttpClientMetricsQueueLength() throws Exception {
     assertWaitUntil(() -> requests.size() == 3);
     assertEquals(Collections.singleton("localhost:8080"), metrics.endpoints());
     assertEquals(0, (int)metrics.queueSize("localhost:8080"));
-    assertEquals(5, (int)metrics.connectionCount("localhost:8080"));
+    assertEquals(3, (int)metrics.connectionCount("localhost:8080"));
   }
 
   @Test
diff --git a/src/test/java/io/vertx/test/core/net/ConnectionPoolTest.java b/src/test/java/io/vertx/test/core/net/ConnectionPoolTest.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/test/core/net/ConnectionPoolTest.java
@@ -0,0 +1,775 @@
+/*
+ * Copyright (c) 2011-2014 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ *     The Eclipse Public License is available at
+ *     http://www.eclipse.org/legal/epl-v10.html
+ *
+ *     The Apache License v2.0 is available at
+ *     http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.vertx.test.core.net;
+
+import io.netty.channel.Channel;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.vertx.core.Vertx;
+import io.vertx.core.http.impl.pool.*;
+import io.vertx.core.impl.ContextImpl;
+import io.vertx.core.impl.ContextInternal;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.test.core.VertxTestBase;
+import org.junit.Test;
+
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ */
+public class ConnectionPoolTest extends VertxTestBase {
+
+  class FakeConnectionManager {
+
+    private final FakeConnectionProvider connector;
+    private final int queueMaxSize;
+    private final int maxPoolSize;
+    private Pool<FakeConnection> pool;
+    private Set<FakeConnection> active = new HashSet<>();
+    private boolean closed = true;
+    private int seq;
+
+    FakeConnectionManager(int queueMaxSize, int maxPoolSize, FakeConnectionProvider connector) {
+      this.queueMaxSize = queueMaxSize;
+      this.maxPoolSize = maxPoolSize;
+      this.connector = connector;
+    }
+
+    synchronized int sequence() {
+      return seq;
+    }
+
+    synchronized boolean closed() {
+      return closed;
+    }
+
+    synchronized boolean contains(FakeConnection conn) {
+      return active.contains(conn);
+    }
+
+    synchronized int size() {
+      return active.size();
+    }
+
+    synchronized Pool<FakeConnection> pool() {
+      return pool;
+    }
+
+    void getConnection(Waiter<FakeConnection> waiter) {
+      synchronized (this) {
+        if (closed) {
+          seq++;
+          closed = false;
+          pool = new Pool<>(
+            connector,
+            queueMaxSize,
+            maxPoolSize,
+            v -> {
+              synchronized (FakeConnectionManager.this) {
+                closed = true;
+              }
+            }, (channel, conn) -> {
+            synchronized (FakeConnectionManager.this) {
+              active.add(conn);
+            }
+          }, (channel, conn) -> {
+            synchronized (FakeConnectionManager.this) {
+              active.remove(conn);
+            }
+          }
+          );
+        }
+      }
+      pool.getConnection(waiter);
+    }
+  }
+
+  @Test
+  public void testConnectSuccess() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 4, connector);
+    AtomicReference<Boolean> initLock = new AtomicReference<>();
+    AtomicReference<Boolean> handleLock = new AtomicReference<>();
+    FakeWaiter waiter = new FakeWaiter() {
+      @Override
+      public synchronized void initConnection(ContextInternal ctx, FakeConnection conn) {
+        assertNull(Vertx.currentContext());
+        assertSame(ctx, context);
+        Pool<FakeConnection> pool = mgr.pool();
+        initLock.set(Thread.holdsLock(pool));
+        super.initConnection(ctx, conn);
+      }
+      @Override
+      public synchronized boolean handleConnection(ContextInternal ctx, FakeConnection conn) throws Exception {
+        assertNull(Vertx.currentContext());
+        assertSame(ctx, context);
+        Pool<FakeConnection> pool = mgr.pool();
+        handleLock.set(Thread.holdsLock(pool));
+        return super.handleConnection(ctx, conn);
+      }
+    };
+    mgr.getConnection(waiter);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    assertWaitUntil(waiter::isComplete);
+    assertEquals(Boolean.FALSE, handleLock.get());
+    assertEquals(Boolean.FALSE, initLock.get());
+    waiter.assertInitialized(conn);
+    waiter.assertSuccess(conn);
+    waiter.recycle();
+    assertEquals(0, mgr.size());
+    assertTrue(mgr.closed());
+  }
+
+  @Test
+  public void testConnectFailure() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 4, connector);
+    AtomicReference<Boolean> holdsLock = new AtomicReference<>();
+    FakeWaiter waiter = new FakeWaiter() {
+      @Override
+      public synchronized void handleFailure(ContextInternal ctx, Throwable failure) {
+        assertNull(Vertx.currentContext());
+        assertSame(ctx, context);
+        Pool<FakeConnection> pool = mgr.pool();
+        holdsLock.set(Thread.holdsLock(pool));
+        super.handleFailure(ctx, failure);
+      }
+    };
+    mgr.getConnection(waiter);
+    FakeConnection conn = connector.assertRequest();
+    Throwable failure = new Throwable();
+    conn.fail(failure);
+    assertWaitUntil(waiter::isComplete);
+    assertEquals(Boolean.FALSE, holdsLock.get());
+    waiter.assertNotInitialized();
+    waiter.assertFailure(failure);
+  }
+
+  @Test
+  public void testConnectPoolEmptyWaiterCancelledAfterConnectRequest() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 3, connector);
+    FakeWaiter waiter = new FakeWaiter();
+    mgr.getConnection(waiter);
+    FakeConnection conn = connector.assertRequest();
+    waiter.cancel();
+    conn.connect();
+    waitUntil(() -> mgr.size() == 1);
+    waiter.assertInitialized(conn);
+    assertTrue(waiter.isComplete());
+    assertFalse(waiter.isSuccess());
+    assertFalse(waiter.isFailure());
+    assertTrue(mgr.contains(conn));
+  }
+
+  @Test
+  public void testConnectionFailure() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 3, connector);
+    FakeWaiter waiter = new FakeWaiter();
+    mgr.getConnection(waiter);
+    FakeConnection conn = connector.assertRequest();
+    Exception expected = new Exception();
+    conn.fail(expected);
+    assertWaitUntil(waiter::isComplete);
+    waiter.assertFailure(expected);
+    assertTrue(waiter.isFailure());
+    assertTrue(mgr.closed());
+  }
+
+  @Test
+  public void testRecycleConnection() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 1, connector);
+    FakeWaiter waiter1 = new FakeWaiter();
+    mgr.getConnection(waiter1);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    assertWaitUntil(waiter1::isComplete);
+    FakeWaiter waiter2 = new FakeWaiter();
+    mgr.getConnection(waiter2);
+    connector.assertRequests(0);
+    waiter1.recycle();
+    assertWaitUntil(waiter2::isComplete);
+    waiter2.assertSuccess(conn);
+  }
+
+  @Test
+  public void testRecycleDiscardedConnection() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 1, connector);
+    FakeWaiter waiter1 = new FakeWaiter();
+    mgr.getConnection(waiter1);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    waitUntil(waiter1::isComplete);
+    FakeWaiter waiter2 = new FakeWaiter();
+    mgr.getConnection(waiter2);
+    conn.close();
+    waiter1.recycle();
+    waitUntil(() -> connector.requests() == 1);
+    assertFalse(mgr.closed());
+    FakeConnection conn2 = connector.assertRequest();
+    conn2.connect();
+    waitUntil(waiter2::isSuccess);
+  }
+
+  @Test
+  public void testWaiterThrowsException() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 1, connector);
+    Exception failure = new Exception();
+    FakeWaiter waiter = new FakeWaiter() {
+      @Override
+      public synchronized boolean handleConnection(ContextInternal ctx, FakeConnection conn) throws Exception {
+        throw failure;
+      }
+    };
+    mgr.getConnection(waiter);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    assertEquals(0, mgr.size());
+  }
+
+  @Test
+  public void testEndpointLifecycle() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 1, connector);
+    FakeWaiter waiter1 = new FakeWaiter();
+    mgr.getConnection(waiter1);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    waitUntil(waiter1::isSuccess);
+    conn.close();
+    waitUntil(mgr::closed);
+    FakeWaiter waiter2 = new FakeWaiter();
+    mgr.getConnection(waiter2);
+    assertEquals(2, mgr.sequence());
+  }
+
+  @Test
+  public void testDontCloseEndpointWithInflightRequest() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(3, 2, connector);
+    FakeWaiter waiter1 = new FakeWaiter();
+    mgr.getConnection(waiter1);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    waitUntil(waiter1::isComplete);
+    FakeWaiter waiter2 = new FakeWaiter();
+    mgr.getConnection(waiter2);
+    conn.close();
+    waitUntil(() -> !mgr.contains(conn));
+    assertFalse(mgr.closed());
+  }
+
+  @Test
+  public void testInitialConcurrency() {
+    int n = 10;
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(-1, 1, connector);
+    List<FakeWaiter> waiters = new ArrayList<>();
+    for (int i = 0; i < n; i++) {
+      FakeWaiter waiter = new FakeWaiter();
+      mgr.getConnection(waiter);
+      waiters.add(waiter);
+    }
+    FakeConnection conn = connector.assertRequest();
+    conn.concurrency(n).connect();
+    waiters.forEach(waiter -> {
+      waitUntil(waiter::isSuccess);
+    });
+    waiters.forEach(FakeWaiter::recycle);
+  }
+
+  @Test
+  public void testInitialNoConcurrency() {
+    int n = 10;
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(-1, 1, connector);
+    List<FakeWaiter> waiters = new ArrayList<>();
+    for (int i = 0; i < n; i++) {
+      FakeWaiter waiter = new FakeWaiter();
+      mgr.getConnection(waiter);
+      waiters.add(waiter);
+    }
+    FakeConnection conn = connector.assertRequest();
+    conn.concurrency(0).connect().awaitConnected();
+    conn.concurrency(n - 1);
+    waitUntil(() -> waiters.stream().filter(FakeWaiter::isSuccess).count() == n - 1);
+    waiters.stream().filter(FakeWaiter::isSuccess).findFirst().get().recycle();
+    waiters.forEach(waiter -> {
+      waitUntil(waiter::isSuccess);
+    });
+  }
+
+  @Test
+  public void testRecycleWithoutDispose() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(-1, 1, connector);
+    FakeWaiter waiter1 = new FakeWaiter();
+    mgr.getConnection(waiter1);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    waitUntil(waiter1::isSuccess);
+    conn.recycle(false);
+    FakeWaiter waiter2 = new FakeWaiter();
+    mgr.getConnection(waiter2);
+    waitUntil(waiter1::isSuccess);
+    waiter2.assertSuccess(conn);
+    conn.recycle(true);
+    assertEquals(0, mgr.size());
+  }
+
+  @Test
+  public void testDiscardWaiterWhenFull() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(2, 1, connector);
+    FakeWaiter waiter1 = new FakeWaiter();
+    mgr.getConnection(waiter1);
+    FakeConnection conn = connector.assertRequest();
+    FakeWaiter waiter2 = new FakeWaiter();
+    mgr.getConnection(waiter2);
+    FakeWaiter waiter3 = new FakeWaiter();
+    mgr.getConnection(waiter3);
+    FakeWaiter waiter4 = new FakeWaiter();
+    mgr.getConnection(waiter4);
+    assertWaitUntil(waiter4::isFailure); // Full
+  }
+
+  @Test
+  public void testDiscardConnectionDuringInit() {
+    FakeConnectionProvider connector = new FakeConnectionProvider();
+    FakeConnectionManager mgr = new FakeConnectionManager(2, 1, connector);
+    FakeWaiter waiter1 = new FakeWaiter() {
+      @Override
+      public synchronized void initConnection(ContextInternal ctx, FakeConnection conn) {
+        super.initConnection(ctx, conn);
+        conn.close(); // Close during init
+      }
+    };
+    mgr.getConnection(waiter1);
+    FakeConnection conn = connector.assertRequest();
+    conn.connect();
+    assertWaitUntil(() -> connector.requests() == 1); // Connection close during init - reattempt to connect
+    assertFalse(mgr.closed());
+  }
+
+  @Test
+  public void testStress() {
+    int numActors = 16;
+    int numConnections = 1000;
+
+    FakeConnectionProvider connector = new FakeConnectionProvider() {
+      @Override
+      public long connect(ConnectionListener<FakeConnection> listener, ContextImpl context) {
+        int i = ThreadLocalRandom.current().nextInt(100);
+        FakeConnection conn = new FakeConnection(context, listener);
+        if (i < 10) {
+          conn.fail(new Exception("Could not connect"));
+        } else {
+          conn.connect();
+        }
+        return 1;
+      }
+    };
+    FakeConnectionManager mgr = new FakeConnectionManager(-1, 16, connector);
+
+    Thread[] actors = new Thread[numActors];
+    for (int i = 0; i < numActors; i++) {
+      actors[i] = new Thread(() -> {
+        CountDownLatch latch = new CountDownLatch(numConnections);
+        for (int i1 = 0; i1 < numConnections; i1++) {
+          mgr.getConnection(new Waiter<FakeConnection>((ContextImpl) vertx.getOrCreateContext()) {
+            @Override
+            public void handleFailure(ContextInternal ctx, Throwable failure) {
+              latch.countDown();
+            }
+
+            @Override
+            public void initConnection(ContextInternal ctx, FakeConnection conn) {
+            }
+
+            @Override
+            public boolean handleConnection(ContextInternal ctx, FakeConnection conn) throws Exception {
+              int action = ThreadLocalRandom.current().nextInt(100);
+              if (action < -1) {
+                latch.countDown();
+                return false;
+              } /* else if (i < 30) {
+                latch.countDown();
+                throw new Exception();
+              } */ else {
+                vertx.setTimer(10, id -> {
+                  if (action < 15) {
+                    conn.close();
+                  } else {
+                    conn.recycle();
+                  }
+                  latch.countDown();
+                });
+                return true;
+              }
+            }
+          });
+        }
+        try {
+          latch.await();
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+        }
+      });
+      actors[i].start();
+    }
+
+    for (int i = 0; i < actors.length; i++) {
+      try {
+        actors[i].join();
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }
+
+    assertWaitUntil(() -> mgr.closed());
+
+    // Check state at the end
+    assertEquals(0, mgr.size());
+    assertEquals(0, mgr.pool.waitersCount());
+    assertEquals(0, mgr.pool.waitersInQueue());
+    assertEquals(0, mgr.pool.weight());
+    assertEquals(0, mgr.pool.capacity());
+  }
+
+  class FakeWaiter extends Waiter<FakeConnection> {
+
+    private FakeConnection init;
+    private boolean cancelled;
+    private boolean completed;
+    private Object result;
+
+    FakeWaiter() {
+      super((ContextImpl) vertx.getOrCreateContext());
+    }
+
+    synchronized boolean cancel() {
+      if (completed) {
+        return false;
+      } else {
+        cancelled = true;
+        return true;
+      }
+    }
+
+    synchronized void assertInitialized(FakeConnection conn) {
+      assertSame(conn, init);
+    }
+
+    synchronized void assertNotInitialized() {
+      assertSame(null, init);
+    }
+    synchronized void assertSuccess(FakeConnection conn) {
+      assertSame(conn, result);
+    }
+
+    synchronized void assertFailure(Throwable failure) {
+      assertSame(failure, result);
+    }
+
+    synchronized boolean isComplete() {
+      return completed;
+    }
+
+    synchronized boolean isSuccess() {
+      return completed && result instanceof FakeConnection;
+    }
+
+    synchronized boolean isFailure() {
+      return completed && result instanceof Throwable;
+    }
+
+    @Override
+    public synchronized void handleFailure(ContextInternal ctx, Throwable failure) {
+      assertFalse(completed);
+      completed = true;
+      result = failure;
+    }
+
+    @Override
+    public synchronized void initConnection(ContextInternal ctx, FakeConnection conn) {
+      assertNull(init);
+      assertNotNull(conn);
+      init = conn;
+    }
+
+    @Override
+    public synchronized boolean handleConnection(ContextInternal ctx, FakeConnection conn) throws Exception {
+      assertFalse(completed);
+      completed = true;
+      if (cancelled) {
+        return false;
+      } else {
+        synchronized (conn) {
+          conn.inflight++;
+        }
+        result = conn;
+        return true;
+      }
+    }
+
+    long recycle() {
+      FakeConnection conn = (FakeConnection) result;
+      return conn.recycle();
+    }
+  }
+
+  /*
+    class FakeConnnectionPool implements ConnectionPool<FakeConnection>, Function<SocketAddress, ConnectionPool<FakeConnection>> {
+
+      private final SocketAddress address = SocketAddress.inetSocketAddress(8080, "localhost");
+      private final int maxSize;
+      private final ArrayDeque<FakeConnection> available = new ArrayDeque<>();
+      private final Set<FakeConnection> all = new HashSet<>();
+      private boolean closed = true;
+      private int sequence;
+
+      FakeConnnectionPool(int maxSize) {
+        this.maxSize = maxSize;
+      }
+
+      @Override
+      public Deque<FakeConnection> available() {
+        return available;
+      }
+
+      @Override
+      public Set<FakeConnection> all() {
+        return all;
+      }
+
+      synchronized int size() {
+        return available.size();
+      }
+
+      synchronized boolean contains(FakeConnection conn) {
+        Deque<ConnectionHolder<FakeConnection>> a = (Deque<ConnectionHolder<FakeConnection>>)(Deque) available;
+        for (ConnectionHolder<FakeConnection> b : a) {
+          if (b.connection() == conn) {
+            return true;
+          }
+        }
+        return false;
+      }
+
+      synchronized int sequence() {
+        return sequence;
+      }
+
+      @Override
+      public synchronized FakeConnnectionPool apply(SocketAddress socketAddress) {
+        if (!socketAddress.equals(address)) {
+          throw new AssertionError();
+        }
+        if (!closed) {
+          throw new AssertionError();
+        }
+        closed = false;
+        sequence++;
+        return this;
+      }
+
+      @Override
+      public synchronized int maxSize() {
+        if (closed) {
+          throw new AssertionError();
+        }
+        return maxSize;
+      }
+
+      @Override
+      public synchronized boolean canBorrow(int connCount) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public synchronized FakeConnection pollConnection() {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public synchronized boolean canCreateConnection(int connCount) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public synchronized boolean initConnection(FakeConnection conn) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public synchronized void recycleConnection(FakeConnection conn) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public synchronized void evictConnection(FakeConnection conn) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public synchronized boolean isValid(FakeConnection conn) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public synchronized ContextImpl getContext(FakeConnection conn) {
+        throw new UnsupportedOperationException();
+      }
+
+      public synchronized void close() {
+        if (closed) {
+          throw new AssertionError();
+        }
+        closed = true;
+        available.clear();
+        all.clear();
+      }
+
+      synchronized boolean isClosed() {
+        return closed;
+      }
+    }
+  */
+  class FakeConnection {
+
+    private static final int DISCONNECTED = 0;
+    private static final int CONNECTING = 1;
+    private static final int CONNECTED = 2;
+    private static final int CLOSED = 3;
+
+    private final ContextImpl context;
+    private final ConnectionListener<FakeConnection> listener;
+    private final Channel channel = new EmbeddedChannel();
+
+    private long inflight;
+    private long concurrency = 1;
+    private int status = DISCONNECTED;
+
+    FakeConnection(ContextImpl context, ConnectionListener<FakeConnection> listener) {
+      this.context = context;
+      this.listener = listener;
+    }
+
+    synchronized void close() {
+      if (status != CONNECTED) {
+        throw new IllegalStateException();
+      }
+      status = CLOSED;
+      listener.onDiscard();
+    }
+
+    synchronized long recycle(boolean dispose) {
+      return recycle(1, dispose);
+    }
+
+    synchronized long recycle() {
+      return recycle(true);
+    }
+
+    synchronized long recycle(int capacity, boolean dispose) {
+      inflight -= capacity;
+      listener.onRecycle(capacity, dispose);
+      return inflight;
+    }
+
+    synchronized FakeConnection concurrency(long value) {
+      if (value < 0) {
+        throw new IllegalArgumentException("Invalid concurrency");
+      }
+      if (status == CONNECTED) {
+        if (concurrency != value) {
+          concurrency = value;
+          listener.onConcurrencyChange(value);
+        }
+      } else {
+        concurrency = value;
+      }
+      return this;
+    }
+
+    FakeConnection awaitConnected() {
+      waitUntil(() -> {
+        synchronized (FakeConnection.this) {
+          return status == CONNECTED;
+        }
+      });
+      return this;
+    }
+
+    synchronized FakeConnection connect() {
+      if (status != DISCONNECTED) {
+        throw new IllegalStateException();
+      }
+      status = CONNECTING;
+      context.nettyEventLoop().execute(() -> {
+        synchronized (FakeConnection.this) {
+          status = CONNECTED;
+          listener.onConnectSuccess(this, concurrency, channel, context, 1, 1);
+        }
+      });
+      return this;
+    }
+
+    void fail(Throwable err) {
+      context.nettyEventLoop().execute(() -> listener.onConnectFailure(context, err, 1));
+    }
+  }
+
+  class FakeConnectionProvider implements ConnectionProvider<FakeConnection> {
+
+    private final ArrayDeque<FakeConnection> pendingRequests = new ArrayDeque<>();
+
+    void assertRequests(int expectedSize) {
+      assertEquals(expectedSize, pendingRequests.size());
+    }
+
+    int requests() {
+      return pendingRequests.size();
+    }
+
+    FakeConnection assertRequest() {
+      assertNotNull(pendingRequests);
+      assertTrue(pendingRequests.size() > 0);
+      FakeConnection request = pendingRequests.poll();
+      assertNotNull(request);
+      return request;
+    }
+
+    @Override
+    public long connect(ConnectionListener<FakeConnection> listener, ContextImpl context) {
+      pendingRequests.add(new FakeConnection(context, listener));
+      return 1;
+    }
+
+    @Override
+    public void close(FakeConnection conn) {
+      conn.listener.onDiscard();
+    }
+  }
+}