Repository: eclipse-vertx/vert.x

Problem statement:
HTTP client pool proxy aware
The HTTP client pool discrimines connection by their host (and SSL) since proxy used to be configured at the client level. Since we allow proxy settings on a request, we should handle proxy settings in the client pool, providing dynamic proxy configuration for environment that use rolling proxies.
 


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/http/impl/EndpointKey.java b/src/main/java/io/vertx/core/http/impl/EndpointKey.java
--- a/src/main/java/io/vertx/core/http/impl/EndpointKey.java
+++ b/src/main/java/io/vertx/core/http/impl/EndpointKey.java
@@ -13,6 +13,8 @@
 import io.vertx.core.net.ProxyOptions;
 import io.vertx.core.net.SocketAddress;
 
+import java.util.Objects;
+
 final class EndpointKey {
 
   final boolean ssl;
@@ -38,7 +40,7 @@ public boolean equals(Object o) {
     if (this == o) return true;
     if (o == null || getClass() != o.getClass()) return false;
     EndpointKey that = (EndpointKey) o;
-    return ssl == that.ssl && serverAddr.equals(that.serverAddr) && peerAddr.equals(that.peerAddr);
+    return ssl == that.ssl && serverAddr.equals(that.serverAddr) && peerAddr.equals(that.peerAddr) && equals(proxyOptions, that.proxyOptions);
   }
 
   @Override
@@ -46,6 +48,31 @@ public int hashCode() {
     int result = ssl ? 1 : 0;
     result = 31 * result + peerAddr.hashCode();
     result = 31 * result + serverAddr.hashCode();
+    if (proxyOptions != null) {
+      result = 31 * result + hashCode(proxyOptions);
+    }
     return result;
   }
+
+  private static boolean equals(ProxyOptions options1, ProxyOptions options2) {
+    if (options1 == options2) {
+      return true;
+    }
+    if (options1 != null && options2 != null) {
+      return Objects.equals(options1.getHost(), options2.getHost()) &&
+        options1.getPort() == options2.getPort() &&
+        Objects.equals(options1.getUsername(), options2.getUsername()) &&
+        Objects.equals(options1.getPassword(), options2.getPassword());
+    }
+    return false;
+  }
+
+  private static int hashCode(ProxyOptions options) {
+    if (options.getUsername() != null && options.getPassword() != null) {
+      return Objects.hash(options.getHost(), options.getPort(), options.getType(), options.getUsername(), options.getPassword());
+    } else {
+      Objects.hash(options.getHost(), options.getPort(), options.getType());
+    }
+    return 0;
+  }
 }
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
@@ -25,6 +25,7 @@
 import io.vertx.core.net.impl.pool.ConnectionManager;
 import io.vertx.core.net.impl.pool.ConnectionPool;
 import io.vertx.core.net.impl.pool.Endpoint;
+import io.vertx.core.net.impl.pool.EndpointProvider;
 import io.vertx.core.net.impl.pool.Lease;
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
@@ -216,28 +217,11 @@ private void checkExpired(Handler<Long> checker) {
   }
 
   private ConnectionManager<EndpointKey, Lease<HttpClientConnection>> httpConnectionManager() {
-    int maxPoolSize = Math.max(options.getMaxPoolSize(), options.getHttp2MaxPoolSize());
-    return new ConnectionManager<>((key, ctx, dispose) -> {
-      ClientMetrics metrics = this.metrics != null ? this.metrics.createEndpointMetrics(key.serverAddr, maxPoolSize) : null;
-      HttpChannelConnector connector = new HttpChannelConnector(this, netClient, key.proxyOptions, metrics, options.getProtocolVersion(), key.ssl, options.isUseAlpn(), key.peerAddr, key.serverAddr);
-      return new SharedClientHttpStreamEndpoint(
-        this,
-        metrics,
-        options.getMaxWaitQueueSize(),
-        options.getMaxPoolSize(),
-        options.getHttp2MaxPoolSize(),
-        connector,
-        dispose);
-    });
+    return new ConnectionManager<>();
   }
 
   private ConnectionManager<EndpointKey, HttpClientConnection> webSocketConnectionManager() {
-    int maxPoolSize = options.getMaxWebSockets();
-    return new ConnectionManager<>((key, ctx, dispose) -> {
-      ClientMetrics metrics = this.metrics != null ? this.metrics.createEndpointMetrics(key.serverAddr, maxPoolSize) : null;
-      HttpChannelConnector connector = new HttpChannelConnector(this, netClient, key.proxyOptions, metrics, HttpVersion.HTTP_1_1, key.ssl, false, key.peerAddr, key.serverAddr);
-      return new WebSocketEndpoint(null, maxPoolSize, connector, dispose);
-    });
+    return new ConnectionManager<>();
   }
 
   Function<ContextInternal, EventLoopContext> contextProvider() {
@@ -275,6 +259,16 @@ private String getHost(RequestOptions request) {
     return options.getDefaultHost();
   }
 
+  private ProxyOptions resolveProxyOptions(ProxyOptions proxyOptions, SocketAddress addr) {
+    proxyOptions = getProxyOptions(proxyOptions);
+    if (proxyFilter != null) {
+      if (!proxyFilter.test(addr)) {
+        proxyOptions = null;
+      }
+    }
+    return proxyOptions;
+  }
+
   HttpClientMetrics metrics() {
     return metrics;
   }
@@ -303,15 +297,10 @@ public void webSocket(WebSocketConnectOptions connectOptions, Handler<AsyncResul
   }
 
   private void webSocket(WebSocketConnectOptions connectOptions, PromiseInternal<WebSocket> promise) {
-    ProxyOptions proxyOptions = getProxyOptions(connectOptions.getProxyOptions());
     int port = getPort(connectOptions);
     String host = getHost(connectOptions);
     SocketAddress addr = SocketAddress.inetSocketAddress(port, host);
-    if (proxyFilter != null) {
-      if (!proxyFilter.test(addr)) {
-        proxyOptions = null;
-      }
-    }
+    ProxyOptions proxyOptions = resolveProxyOptions(connectOptions.getProxyOptions(), addr);
     EndpointKey key = new EndpointKey(connectOptions.isSsl() != null ? connectOptions.isSsl() : options.isSsl(), proxyOptions, addr, addr);
     ContextInternal ctx = promise.context();
     EventLoopContext eventLoopContext;
@@ -320,9 +309,19 @@ private void webSocket(WebSocketConnectOptions connectOptions, PromiseInternal<W
     } else {
       eventLoopContext = vertx.createEventLoopContext(ctx.nettyEventLoop(), ctx.workerPool(), ctx.classLoader());
     }
+    EndpointProvider<HttpClientConnection> provider = new EndpointProvider<HttpClientConnection>() {
+      @Override
+      public Endpoint<HttpClientConnection> create(ContextInternal ctx, Runnable dispose) {
+        int maxPoolSize = options.getMaxWebSockets();
+        ClientMetrics metrics = HttpClientImpl.this.metrics != null ? HttpClientImpl.this.metrics.createEndpointMetrics(key.serverAddr, maxPoolSize) : null;
+        HttpChannelConnector connector = new HttpChannelConnector(HttpClientImpl.this, netClient, proxyOptions, metrics, HttpVersion.HTTP_1_1, key.ssl, false, key.peerAddr, key.serverAddr);
+        return new WebSocketEndpoint(null, maxPoolSize, connector, dispose);
+      }
+    };
     webSocketCM.getConnection(
       eventLoopContext,
       key,
+      provider,
       ar -> {
         if (ar.succeeded()) {
           Http1xClientConnection conn = (Http1xClientConnection) ar.result();
@@ -550,7 +549,6 @@ private void doRequest(RequestOptions request, PromiseInternal<HttpClientRequest
     if (server == null) {
       server = SocketAddress.inetSocketAddress(port, host);
     }
-    ProxyOptions proxyOptions = getProxyOptions(request.getProxyOptions());
     HttpMethod method = request.getMethod();
     String requestURI = request.getURI();
     Boolean ssl = request.isSsl();
@@ -566,37 +564,36 @@ private void doRequest(RequestOptions request, PromiseInternal<HttpClientRequest
       throw new IllegalArgumentException("Must enable ALPN when using H2");
     }
     checkClosed();
-    if (proxyFilter != null) {
-      if (!proxyFilter.test(server)) {
-        proxyOptions = null;
-      }
-    }
-    if (proxyOptions != null) {
-      if (!useSSL && proxyOptions.getType() == ProxyType.HTTP) {
-        // If the requestURI is as not absolute URI then we do not recompute one for the proxy
-        if (!ABS_URI_START_PATTERN.matcher(requestURI).find()) {
-          int defaultPort = 80;
-          String addPort = (port != -1 && port != defaultPort) ? (":" + port) : "";
-          requestURI = (ssl == Boolean.TRUE ? "https://" : "http://") + host + addPort + requestURI;
-        }
-        if (proxyOptions.getUsername() != null && proxyOptions.getPassword() != null) {
-          if (headers == null) {
-            headers = HttpHeaders.headers();
-          }
-          headers.add("Proxy-Authorization", "Basic " + Base64.getEncoder()
-            .encodeToString((proxyOptions.getUsername() + ":" + proxyOptions.getPassword()).getBytes()));
-        }
-        server = SocketAddress.inetSocketAddress(proxyOptions.getPort(), proxyOptions.getHost());
-        proxyOptions = null;
-      }
-    }
+    ProxyOptions proxyOptions = resolveProxyOptions(request.getProxyOptions(), server);
 
     String peerHost = host;
     if (peerHost.endsWith(".")) {
       peerHost = peerHost.substring(0, peerHost.length() -  1);
     }
     SocketAddress peerAddress = SocketAddress.inetSocketAddress(port, peerHost);
-    doRequest(method, peerAddress, server, host, port, useSSL, requestURI, headers, request.getTraceOperation(), timeout, followRedirects, proxyOptions, promise);
+
+    EndpointKey key;
+    if (proxyOptions != null && !useSSL && proxyOptions.getType() == ProxyType.HTTP) {
+      // If the requestURI is as not absolute URI then we do not recompute one for the proxy
+      if (!ABS_URI_START_PATTERN.matcher(requestURI).find()) {
+        int defaultPort = 80;
+        String addPort = (port != -1 && port != defaultPort) ? (":" + port) : "";
+        requestURI = (ssl == Boolean.TRUE ? "https://" : "http://") + host + addPort + requestURI;
+      }
+      if (proxyOptions.getUsername() != null && proxyOptions.getPassword() != null) {
+        if (headers == null) {
+          headers = HttpHeaders.headers();
+        }
+        headers.add("Proxy-Authorization", "Basic " + Base64.getEncoder()
+          .encodeToString((proxyOptions.getUsername() + ":" + proxyOptions.getPassword()).getBytes()));
+      }
+      server = SocketAddress.inetSocketAddress(proxyOptions.getPort(), proxyOptions.getHost());
+      key = new EndpointKey(useSSL, proxyOptions, server, peerAddress);
+      proxyOptions = null;
+    } else {
+      key = new EndpointKey(useSSL, proxyOptions, server, peerAddress);
+    }
+    doRequest(method, peerAddress, server, host, port, useSSL, requestURI, headers, request.getTraceOperation(), timeout, followRedirects, proxyOptions, key, promise);
   }
 
   private void doRequest(
@@ -612,10 +609,26 @@ private void doRequest(
     long timeout,
     Boolean followRedirects,
     ProxyOptions proxyOptions,
+    EndpointKey key,
     PromiseInternal<HttpClientRequest> requestPromise) {
     ContextInternal ctx = requestPromise.context();
-    EndpointKey key = new EndpointKey(useSSL, proxyOptions, server, peerAddress);
-    httpCM.getConnection(ctx, key, timeout, ar1 -> {
+    EndpointProvider<Lease<HttpClientConnection>> provider = new EndpointProvider<Lease<HttpClientConnection>>() {
+      @Override
+      public Endpoint<Lease<HttpClientConnection>> create(ContextInternal ctx, Runnable dispose) {
+        int maxPoolSize = Math.max(options.getMaxPoolSize(), options.getHttp2MaxPoolSize());
+        ClientMetrics metrics = HttpClientImpl.this.metrics != null ? HttpClientImpl.this.metrics.createEndpointMetrics(key.serverAddr, maxPoolSize) : null;
+        HttpChannelConnector connector = new HttpChannelConnector(HttpClientImpl.this, netClient, proxyOptions, metrics, options.getProtocolVersion(), key.ssl, options.isUseAlpn(), key.peerAddr, key.serverAddr);
+        return new SharedClientHttpStreamEndpoint(
+          HttpClientImpl.this,
+          metrics,
+          options.getMaxWaitQueueSize(),
+          options.getMaxPoolSize(),
+          options.getHttp2MaxPoolSize(),
+          connector,
+          dispose);
+      }
+    };
+    httpCM.getConnection(ctx, key, provider, timeout, ar1 -> {
       if (ar1.succeeded()) {
         Lease<HttpClientConnection> lease = ar1.result();
         HttpClientConnection conn = lease.get();
diff --git a/src/main/java/io/vertx/core/net/impl/pool/ConnectionManager.java b/src/main/java/io/vertx/core/net/impl/pool/ConnectionManager.java
--- a/src/main/java/io/vertx/core/net/impl/pool/ConnectionManager.java
+++ b/src/main/java/io/vertx/core/net/impl/pool/ConnectionManager.java
@@ -14,7 +14,6 @@
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
 import io.vertx.core.impl.ContextInternal;
-import io.vertx.core.impl.EventLoopContext;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -28,30 +27,27 @@
  */
 public class ConnectionManager<K, C> {
 
-  private final EndpointProvider<K, C> endpointProvider;
   private final Map<K, Endpoint<C>> endpointMap = new ConcurrentHashMap<>();
 
-  public ConnectionManager(EndpointProvider<K, C> endpointProvider) {
-    this.endpointProvider = endpointProvider;
-  }
-
   public void forEach(Consumer<Endpoint<C>> consumer) {
     endpointMap.values().forEach(consumer);
   }
 
   public void getConnection(ContextInternal ctx,
                             K key,
+                            EndpointProvider<C> provider,
                             Handler<AsyncResult<C>> handler) {
-    getConnection(ctx, key, 0, handler);
+    getConnection(ctx, key, provider, 0, handler);
   }
 
   public void getConnection(ContextInternal ctx,
                             K key,
+                            EndpointProvider<C> provider,
                             long timeout,
                             Handler<AsyncResult<C>> handler) {
     Runnable dispose = () -> endpointMap.remove(key);
     while (true) {
-      Endpoint<C> endpoint = endpointMap.computeIfAbsent(key, k -> endpointProvider.create(key, ctx, dispose));
+      Endpoint<C> endpoint = endpointMap.computeIfAbsent(key, k -> provider.create(ctx, dispose));
       if (endpoint.getConnection(ctx, timeout, handler)) {
         break;
       }
diff --git a/src/main/java/io/vertx/core/net/impl/pool/EndpointProvider.java b/src/main/java/io/vertx/core/net/impl/pool/EndpointProvider.java
--- a/src/main/java/io/vertx/core/net/impl/pool/EndpointProvider.java
+++ b/src/main/java/io/vertx/core/net/impl/pool/EndpointProvider.java
@@ -17,15 +17,15 @@
  *
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-public interface EndpointProvider<K, C> {
+public interface EndpointProvider<C> {
 
   /**
    * Create an endpoint tracked by the {@link ConnectionManager}.
    *
+   * @param ctx     the creating context
    * @param dispose the callback to signal this endpoint should be destroyed
-   * @param ctx the creating context
    * @return the created endpoint
    */
-  Endpoint<C> create(K key, ContextInternal ctx, Runnable dispose);
+  Endpoint<C> create(ContextInternal ctx, Runnable dispose);
 
 }


Test patch:
diff --git a/src/test/java/io/vertx/core/http/Http1xProxyTest.java b/src/test/java/io/vertx/core/http/Http1xProxyTest.java
--- a/src/test/java/io/vertx/core/http/Http1xProxyTest.java
+++ b/src/test/java/io/vertx/core/http/Http1xProxyTest.java
@@ -17,13 +17,19 @@
 import io.vertx.core.net.ProxyOptions;
 import io.vertx.core.net.ProxyType;
 import io.vertx.core.net.SocketAddress;
+import io.vertx.test.proxy.HttpProxy;
+import io.vertx.test.proxy.SocksProxy;
+import io.vertx.test.proxy.TestProxyBase;
 import io.vertx.test.tls.Cert;
 import org.junit.Test;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 public class Http1xProxyTest extends HttpTestBase {
@@ -46,7 +52,7 @@ public void testHttpProxyRequest() throws Exception {
     startProxy(null, ProxyType.HTTP);
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.getPort())));
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port())));
     testHttpProxyRequest(() -> client.request(new RequestOptions()
       .setHost(DEFAULT_HTTP_HOST)
       .setPort(DEFAULT_HTTP_PORT)
@@ -62,7 +68,7 @@ public void testHttpProxyRequest() throws Exception {
   public void testHttpProxyRequest2() throws Exception {
     startProxy(null, ProxyType.HTTP);
     testHttpProxyRequest(() -> client.request(new RequestOptions()
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.getPort()))
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port()))
       .setHost(DEFAULT_HTTP_HOST)
       .setPort(DEFAULT_HTTP_PORT)
       .setURI("/")
@@ -87,7 +93,7 @@ private void testFilter(boolean accept) throws Exception {
     startProxy(null, ProxyType.HTTP);
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.getPort())));
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port())));
     Set<SocketAddress> filtered = Collections.synchronizedSet(new HashSet<>());
     ((HttpClientImpl)client).proxyFilter(so -> {
       filtered.add(so);
@@ -132,7 +138,7 @@ private void testNonProxyHosts(List<String> nonProxyHosts, String host, boolean
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
       .setNonProxyHosts(nonProxyHosts)
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.getPort())));
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port())));
     testHttpProxyRequest(() -> client.request(new RequestOptions()
       .setHost(host)
       .setPort(DEFAULT_HTTP_PORT)
@@ -151,7 +157,7 @@ public void testHttpProxyRequestOverrideClientSsl() throws Exception {
     startProxy(null, ProxyType.HTTP);
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
-      .setSsl(true).setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.getPort())));
+      .setSsl(true).setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port())));
     testHttpProxyRequest(() -> client
       .request(new RequestOptions().setSsl(false).setHost("localhost").setPort(8080))
       .compose(HttpClientRequest::send)).onComplete(onSuccess(v -> {
@@ -189,7 +195,7 @@ public void testHttpProxyRequestAuth() throws Exception {
 
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.getPort())
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port())
         .setUsername("user").setPassword("user")));
 
     server.requestHandler(req -> {
@@ -218,7 +224,7 @@ public void testHttpProxyFtpRequest() throws Exception {
     startProxy(null, ProxyType.HTTP);
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.getPort())));
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port())));
     final String url = "ftp://ftp.gnu.org/gnu/";
     proxy.setForceUri("http://localhost:8080/");
     server.requestHandler(req -> {
@@ -244,7 +250,7 @@ public void testHttpSocksProxyRequest() throws Exception {
 
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5).setHost("localhost").setPort(proxy.getPort())));
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5).setHost("localhost").setPort(proxy.port())));
 
     server.requestHandler(req -> req.response().end());
 
@@ -269,7 +275,7 @@ public void testHttpSocksProxyRequestAuth() throws Exception {
 
     client.close();
     client = vertx.createHttpClient(new HttpClientOptions()
-      .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5).setHost("localhost").setPort(proxy.getPort())
+      .setProxyOptions(new ProxyOptions().setType(ProxyType.SOCKS5).setHost("localhost").setPort(proxy.port())
         .setUsername("user").setPassword("user")));
 
     server.requestHandler(req -> {
@@ -291,6 +297,178 @@ public void testHttpSocksProxyRequestAuth() throws Exception {
     await();
   }
 
+  @Test
+  public void testHttpProxyPooling() throws Exception {
+    HttpProxy proxy1 = new HttpProxy().port(HttpProxy.DEFAULT_PORT);
+    HttpProxy proxy2 = new HttpProxy().port(HttpProxy.DEFAULT_PORT + 1);
+    ProxyOptions req1 = new ProxyOptions()
+      .setType(ProxyType.HTTP)
+      .setHost("localhost")
+      .setPort(proxy1.port());
+    ProxyOptions req2 = new ProxyOptions()
+      .setType(ProxyType.HTTP)
+      .setHost("localhost")
+      .setPort(proxy2.port());
+    List<String> res = testPooling(req1, req2, proxy1, proxy2);
+    assertEquals(Arrays.asList(proxy1.lastLocalAddress(), proxy2.lastLocalAddress()), res);
+  }
+
+  @Test
+  public void testHttpProxyPooling2() throws Exception {
+    HttpProxy proxy = new HttpProxy().port(HttpProxy.DEFAULT_PORT);
+    ProxyOptions req = new ProxyOptions()
+      .setType(ProxyType.HTTP)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    List<String> res = testPooling(req, req, proxy);
+    assertEquals(Arrays.asList(proxy.lastLocalAddress(), proxy.lastLocalAddress()), res);
+  }
+
+  @Test
+  public void testHttpProxyAuthPooling1() throws Exception {
+    HttpProxy proxy = new HttpProxy().port(SocksProxy.DEFAULT_PORT).username(Arrays.asList("user1", "user2"));
+    ProxyOptions req1 = new ProxyOptions()
+      .setUsername("user1")
+      .setPassword("user1")
+      .setType(ProxyType.HTTP)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    ProxyOptions req2 = new ProxyOptions()
+      .setUsername("user2")
+      .setPassword("user2")
+      .setType(ProxyType.HTTP)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    List<String> res = testPooling(req1, req2, proxy);
+    assertEquals(proxy.localAddresses(), res);
+  }
+
+  @Test
+  public void testHttpProxyAuthPooling2() throws Exception {
+    HttpProxy proxy = new HttpProxy().port(SocksProxy.DEFAULT_PORT).username(Arrays.asList("user1"));
+    ProxyOptions req1 = new ProxyOptions()
+      .setUsername("user1")
+      .setPassword("user1")
+      .setType(ProxyType.HTTP)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    ProxyOptions req2 = new ProxyOptions()
+      .setUsername("user1")
+      .setPassword("user1")
+      .setType(ProxyType.HTTP)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    List<String> res = testPooling(req1, req2, proxy);
+    assertEquals(1, proxy.localAddresses().size());
+    assertEquals(Arrays.asList(proxy.localAddresses().get(0), proxy.localAddresses().get(0)), res);
+  }
+
+  @Test
+  public void testSocksProxyPooling1() throws Exception {
+    SocksProxy proxy1 = new SocksProxy().port(SocksProxy.DEFAULT_PORT);
+    SocksProxy proxy2 = new SocksProxy().port(SocksProxy.DEFAULT_PORT + 1);
+    ProxyOptions req1 = new ProxyOptions()
+      .setType(ProxyType.SOCKS5)
+      .setHost("localhost")
+      .setPort(proxy1.port());
+    ProxyOptions req2 = new ProxyOptions()
+      .setType(ProxyType.SOCKS5)
+      .setHost("localhost")
+      .setPort(proxy2.port());
+    List<String> res = testPooling(req1, req2, proxy1, proxy2);
+    assertEquals(Arrays.asList(proxy1.lastLocalAddress(), proxy2.lastLocalAddress()), res);
+  }
+
+  @Test
+  public void testSocksProxyPooling2() throws Exception {
+    SocksProxy proxy = new SocksProxy().port(SocksProxy.DEFAULT_PORT);
+    ProxyOptions req = new ProxyOptions()
+      .setType(ProxyType.SOCKS5)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    List<String> res = testPooling(req, req, proxy);
+    assertEquals(Arrays.asList(proxy.lastLocalAddress(), proxy.lastLocalAddress()), res);
+  }
+
+  @Test
+  public void testSocksProxyAuthPooling1() throws Exception {
+    SocksProxy proxy = new SocksProxy().port(SocksProxy.DEFAULT_PORT).username(Arrays.asList("user1", "user2"));
+    ProxyOptions req1 = new ProxyOptions()
+      .setUsername("user1")
+      .setPassword("user1")
+      .setType(ProxyType.SOCKS5)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    ProxyOptions req2 = new ProxyOptions()
+      .setUsername("user2")
+      .setPassword("user2")
+      .setType(ProxyType.SOCKS5)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    List<String> res = testPooling(req1, req2, proxy);
+    assertEquals(proxy.localAddresses(), res);
+  }
+
+  @Test
+  public void testSocksProxyAuthPooling2() throws Exception {
+    SocksProxy proxy = new SocksProxy().port(SocksProxy.DEFAULT_PORT).username(Arrays.asList("user1"));
+    ProxyOptions req1 = new ProxyOptions()
+      .setUsername("user1")
+      .setPassword("user1")
+      .setType(ProxyType.SOCKS5)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    ProxyOptions req2 = new ProxyOptions()
+      .setUsername("user1")
+      .setPassword("user1")
+      .setType(ProxyType.SOCKS5)
+      .setHost("localhost")
+      .setPort(proxy.port());
+    List<String> res = testPooling(req1, req2, proxy);
+    assertEquals(1, proxy.localAddresses().size());
+    assertEquals(Arrays.asList(proxy.localAddresses().get(0), proxy.localAddresses().get(0)), res);
+  }
+
+  public List<String> testPooling(ProxyOptions request1, ProxyOptions request2, TestProxyBase... proxies) throws Exception {
+    for (TestProxyBase proxy : proxies) {
+      proxy.start(vertx);
+    }
+
+    client.close();
+    client = vertx.createHttpClient(new HttpClientOptions().setMaxPoolSize(1).setKeepAlive(true));
+
+    CompletableFuture<List<String>> ret = new CompletableFuture<>();
+
+    try {
+      server.requestHandler(req -> {
+        SocketAddress addr = req.connection().remoteAddress();
+        req.response().end("" + addr);
+      }).listen(onSuccess(s -> {
+        RequestOptions baseOptions = new RequestOptions()
+          .setHost(DEFAULT_HTTP_HOST)
+          .setPort(DEFAULT_HTTP_PORT)
+          .setURI("/");
+        client.request(new RequestOptions(baseOptions).setProxyOptions(request1))
+          .compose(HttpClientRequest::send)
+          .compose(HttpClientResponse::body)
+          .onComplete(onSuccess(res1 -> {
+            client.request(new RequestOptions(baseOptions).setProxyOptions(request2))
+              .compose(HttpClientRequest::send)
+              .compose(HttpClientResponse::body)
+              .onComplete(onSuccess(res2 -> {
+                ret.complete(Arrays.asList(res1.toString(), res2.toString()));
+              }));
+          }));
+      }));
+
+      return ret.get(40, TimeUnit.SECONDS);
+    } finally {
+      for (TestProxyBase proxy : proxies) {
+        proxy.stop();
+      }
+    }
+  }
+
   @Test
   public void testWssHttpProxy() throws Exception {
     startProxy(null, ProxyType.HTTP);
@@ -301,7 +479,7 @@ public void testWssHttpProxy() throws Exception {
       .setProxyOptions(new ProxyOptions()
         .setType(ProxyType.HTTP)
         .setHost(DEFAULT_HTTP_HOST)
-        .setPort(proxy.getPort())), true);
+        .setPort(proxy.port())), true);
   }
 
   @Test
@@ -311,7 +489,7 @@ public void testWsHttpProxy() throws Exception {
       .setProxyOptions(new ProxyOptions()
         .setType(ProxyType.HTTP)
         .setHost(DEFAULT_HTTP_HOST)
-        .setPort(proxy.getPort())), true);
+        .setPort(proxy.port())), true);
   }
 
   @Test
@@ -324,7 +502,7 @@ public void testWssSocks5Proxy() throws Exception {
       .setProxyOptions(new ProxyOptions()
         .setType(ProxyType.SOCKS5)
         .setHost(DEFAULT_HTTP_HOST)
-        .setPort(proxy.getPort())), true);
+        .setPort(proxy.port())), true);
   }
 
   @Test
@@ -334,7 +512,7 @@ public void testWsSocks5Proxy() throws Exception {
       .setProxyOptions(new ProxyOptions()
         .setType(ProxyType.SOCKS5)
         .setHost(DEFAULT_HTTP_HOST)
-        .setPort(proxy.getPort())), true);
+        .setPort(proxy.port())), true);
   }
 
   @Test
@@ -345,7 +523,7 @@ public void testWsNonProxyHosts() throws Exception {
       .setProxyOptions(new ProxyOptions()
         .setType(ProxyType.HTTP)
         .setHost(DEFAULT_HTTP_HOST)
-        .setPort(proxy.getPort())), false);
+        .setPort(proxy.port())), false);
   }
 
   private void testWebSocket(HttpServerOptions serverOptions, HttpClientOptions clientOptions, boolean proxied) throws Exception {
diff --git a/src/test/java/io/vertx/core/http/HttpTestBase.java b/src/test/java/io/vertx/core/http/HttpTestBase.java
--- a/src/test/java/io/vertx/core/http/HttpTestBase.java
+++ b/src/test/java/io/vertx/core/http/HttpTestBase.java
@@ -138,10 +138,11 @@ protected void startServer(SocketAddress bindAddress, Context context, HttpServe
 
   protected void startProxy(String username, ProxyType proxyType) throws Exception {
     if (proxyType == ProxyType.HTTP) {
-      proxy = new HttpProxy(username);
+      proxy = new HttpProxy();
     } else {
-      proxy = new SocksProxy(username);
+      proxy = new SocksProxy();
     }
+    proxy.username(username);
     proxy.start(vertx);
   }
 }
diff --git a/src/test/java/io/vertx/core/net/NetTest.java b/src/test/java/io/vertx/core/net/NetTest.java
--- a/src/test/java/io/vertx/core/net/NetTest.java
+++ b/src/test/java/io/vertx/core/net/NetTest.java
@@ -2991,7 +2991,7 @@ public void testWithSocks5Proxy() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new SocksProxy(null);
+    proxy = new SocksProxy();
     proxy.start(vertx);
     server.listen(1234, "localhost", ar -> {
       assertTrue(ar.succeeded());
@@ -3020,7 +3020,7 @@ public void testWithSocks5ProxyAuth() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new SocksProxy("username");
+    proxy = new SocksProxy().username("username");
     proxy.start(vertx);
     server.listen(1234, "localhost", ar -> {
       assertTrue(ar.succeeded());
@@ -3054,7 +3054,7 @@ public void testConnectSSLWithSocks5Proxy() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new SocksProxy(null);
+    proxy = new SocksProxy();
     proxy.start(vertx);
     server.listen(ar -> {
       assertTrue(ar.succeeded());
@@ -3088,7 +3088,7 @@ public void testUpgradeSSLWithSocks5Proxy() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new SocksProxy(null);
+    proxy = new SocksProxy();
     proxy.start(vertx);
     server.listen(ar -> {
       assertTrue(ar.succeeded());
@@ -3116,7 +3116,7 @@ public void testWithHttpConnectProxy() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new HttpProxy(null);
+    proxy = new HttpProxy();
     proxy.start(vertx);
     server.listen(1234, "localhost", ar -> {
       assertTrue(ar.succeeded());
@@ -3144,7 +3144,7 @@ public void testWithSocks4aProxy() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new Socks4Proxy(null);
+    proxy = new Socks4Proxy();
     proxy.start(vertx);
     server.listen(1234, "localhost", ar -> {
       assertTrue(ar.succeeded());
@@ -3173,7 +3173,7 @@ public void testWithSocks4aProxyAuth() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new Socks4Proxy("username");
+    proxy = new Socks4Proxy().username("username");
     proxy.start(vertx);
     server.listen(1234, "localhost", ar -> {
       assertTrue(ar.succeeded());
@@ -3201,7 +3201,7 @@ public void testWithSocks4LocalResolver() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new Socks4Proxy(null).start(vertx);
+    proxy = new Socks4Proxy().start(vertx);
     server.listen(1234, "localhost", ar -> {
       assertTrue(ar.succeeded());
       client.connect(1234, "127.0.0.1", ar2 -> {
@@ -3226,7 +3226,7 @@ public void testNonProxyHosts() throws Exception {
     server.connectHandler(sock -> {
 
     });
-    proxy = new HttpProxy(null);
+    proxy = new HttpProxy();
     proxy.start(vertx);
     server.listen(1234, "localhost", onSuccess(s -> {
       client.connect(1234, "example.com", onSuccess(so -> {
diff --git a/src/test/java/io/vertx/core/net/ProxyErrorTest.java b/src/test/java/io/vertx/core/net/ProxyErrorTest.java
--- a/src/test/java/io/vertx/core/net/ProxyErrorTest.java
+++ b/src/test/java/io/vertx/core/net/ProxyErrorTest.java
@@ -69,7 +69,7 @@ protected VertxOptions getOptions() {
   // we don't start http/https servers, due to the error, they will not be queried
 
   private void startProxy(int error, String username) throws Exception {
-    proxy = new HttpProxy(username);
+    proxy = new HttpProxy().username(username);
     proxy.setError(error);
     proxy.start(vertx);
   }
@@ -128,7 +128,7 @@ private void proxyTest(int error, String username, String url, Handler<AsyncResu
         .setProxyOptions(new ProxyOptions()
             .setType(ProxyType.HTTP)
             .setHost("localhost")
-            .setPort(proxy.getPort()));
+            .setPort(proxy.port()));
     HttpClient client = vertx.createHttpClient(options);
 
     client.request(new RequestOptions().setAbsoluteURI(url), ar -> {
diff --git a/src/test/java/io/vertx/core/net/impl/pool/ConnectionManagerTest.java b/src/test/java/io/vertx/core/net/impl/pool/ConnectionManagerTest.java
--- a/src/test/java/io/vertx/core/net/impl/pool/ConnectionManagerTest.java
+++ b/src/test/java/io/vertx/core/net/impl/pool/ConnectionManagerTest.java
@@ -44,9 +44,9 @@ private void testGetConnection(boolean success) {
     EventLoopContext ctx = (EventLoopContext) vertx.getOrCreateContext();
     Connection result = new Connection();
     Throwable failure = new Throwable();
-    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>(new EndpointProvider<Object, Connection>() {
+    EndpointProvider<Connection> provider = new EndpointProvider<Connection>() {
       @Override
-      public Endpoint<Connection> create(Object key, ContextInternal ctx, Runnable dispose) {
+      public Endpoint<Connection> create(ContextInternal ctx, Runnable dispose) {
         return new Endpoint<Connection>(dispose) {
           @Override
           public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncResult<Connection>> handler) {
@@ -59,8 +59,9 @@ public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncRe
           }
         };
       }
-    });
-    mgr.getConnection(ctx, TEST_KEY, ar -> {
+    };
+    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>();
+    mgr.getConnection(ctx, TEST_KEY, provider, ar -> {
       if (ar.succeeded()) {
         assertTrue(success);
         assertSame(result, ar.result());
@@ -87,9 +88,9 @@ private void testDispose(boolean closeConnectionAfterCallback) {
     EventLoopContext ctx = (EventLoopContext) vertx.getOrCreateContext();
     Connection expected = new Connection();
     boolean[] disposed = new boolean[1];
-    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>(new EndpointProvider<Object, Connection>() {
+    EndpointProvider<Connection> provider = new EndpointProvider<Connection>() {
       @Override
-      public Endpoint<Connection> create(Object key, ContextInternal ctx, Runnable dispose) {
+      public Endpoint<Connection> create(ContextInternal ctx, Runnable dispose) {
         return new Endpoint<Connection>(dispose) {
           @Override
           public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncResult<Connection>> handler) {
@@ -106,14 +107,16 @@ public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncRe
               assertTrue(disposed[0]);
             }
           }
+
           @Override
           protected void dispose() {
             disposed[0] = true;
           }
         };
       }
-    });
-    mgr.getConnection(ctx, TEST_KEY, onSuccess(conn -> {
+    };
+    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>();
+    mgr.getConnection(ctx, TEST_KEY, provider, onSuccess(conn -> {
       assertEquals(expected, conn);
     }));
     waitUntil(() -> disposed[0]);
@@ -124,19 +127,21 @@ public void testCloseManager() throws Exception {
     EventLoopContext ctx = (EventLoopContext) vertx.getOrCreateContext();
     Connection expected = new Connection();
     boolean[] disposed = new boolean[1];
-    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>(new EndpointProvider<Object, Connection>() {
+    EndpointProvider<Connection> provider = new EndpointProvider<Connection>() {
       @Override
-      public Endpoint<Connection> create(Object key, ContextInternal ctx, Runnable dispose) {
+      public Endpoint<Connection> create(ContextInternal ctx, Runnable dispose) {
         return new Endpoint<Connection>(dispose) {
           @Override
           public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncResult<Connection>> handler) {
             incRefCount();
             handler.handle(Future.succeededFuture(expected));
           }
+
           @Override
           protected void dispose() {
             disposed[0] = true;
           }
+
           @Override
           protected void close() {
             super.close();
@@ -144,9 +149,10 @@ protected void close() {
           }
         };
       }
-    });
+    };
+    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>();
     CountDownLatch latch = new CountDownLatch(1);
-    mgr.getConnection(ctx, TEST_KEY, onSuccess(conn -> {
+    mgr.getConnection(ctx, TEST_KEY, provider, onSuccess(conn -> {
       assertEquals(expected, conn);
       latch.countDown();
     }));
@@ -162,9 +168,9 @@ public void testCloseManagerImmediately() {
     Connection expected = new Connection();
     boolean[] disposed = new boolean[1];
     AtomicReference<Runnable> adder = new AtomicReference<>();
-    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>(new EndpointProvider<Object, Connection>() {
+    EndpointProvider<Connection> provider = new EndpointProvider<Connection>() {
       @Override
-      public Endpoint<Connection> create(Object key, ContextInternal ctx, Runnable dispose) {
+      public Endpoint<Connection> create(ContextInternal ctx, Runnable dispose) {
         return new Endpoint<Connection>(dispose) {
           @Override
           public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncResult<Connection>> handler) {
@@ -174,8 +180,9 @@ public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncRe
           }
         };
       }
-    });
-    mgr.getConnection(ctx, TEST_KEY, onSuccess(conn -> {
+    };
+    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>();
+    mgr.getConnection(ctx, TEST_KEY, provider, onSuccess(conn -> {
     }));
     waitUntil(() -> adder.get() != null);
     mgr.close();
@@ -186,9 +193,9 @@ public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncRe
   public void testConcurrentDispose() throws Exception {
     EventLoopContext ctx = (EventLoopContext) vertx.getOrCreateContext();
     ConcurrentLinkedQueue<AtomicBoolean> disposals = new ConcurrentLinkedQueue<>();
-    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>(new EndpointProvider<Object, Connection>() {
+    EndpointProvider<Connection> provider = new EndpointProvider<Connection>() {
       @Override
-      public Endpoint<Connection> create(Object key, ContextInternal ctx, Runnable dispose) {
+      public Endpoint<Connection> create(ContextInternal ctx, Runnable dispose) {
         AtomicBoolean disposed = new AtomicBoolean();
         disposals.add(disposed);
         return new Endpoint<Connection>(dispose) {
@@ -204,13 +211,15 @@ public void requestConnection(ContextInternal ctx, long timeout, Handler<AsyncRe
               decRefCount();
             }
           }
+
           @Override
           protected void dispose() {
             disposed.set(true);
           }
         };
       }
-    });
+    };
+    ConnectionManager<Object, Connection> mgr = new ConnectionManager<>();
     int num = 100000;
     int concurrency = 4;
     CountDownLatch[] latches = new CountDownLatch[concurrency];
@@ -219,7 +228,7 @@ protected void dispose() {
       latches[i] = cc;
       new Thread(() -> {
         for (int j = 0;j < num;j++) {
-          mgr.getConnection(ctx, TEST_KEY, onSuccess(conn -> {
+          mgr.getConnection(ctx, TEST_KEY, provider, onSuccess(conn -> {
             cc.countDown();
           }));
         }
diff --git a/src/test/java/io/vertx/test/proxy/HttpProxy.java b/src/test/java/io/vertx/test/proxy/HttpProxy.java
--- a/src/test/java/io/vertx/test/proxy/HttpProxy.java
+++ b/src/test/java/io/vertx/test/proxy/HttpProxy.java
@@ -13,7 +13,9 @@
 
 import java.net.UnknownHostException;
 import java.util.Base64;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
 import io.vertx.core.MultiMap;
@@ -44,21 +46,23 @@
  * <p>
  * @author <a href="http://oss.lehmann.cx/">Alexander Lehmann</a>
  */
-public class HttpProxy extends TestProxyBase {
+public class HttpProxy extends TestProxyBase<HttpProxy> {
 
-  private static final int PORT = 13128;
+  public static final int DEFAULT_PORT = 13128;
 
   private static final Logger log = LoggerFactory.getLogger(HttpProxy.class);
 
   private HttpServer server;
+  private Map<HttpConnection, HttpClient> clientMap = new ConcurrentHashMap<>();
 
   private int error = 0;
 
   private MultiMap lastRequestHeaders = null;
   private HttpMethod lastMethod;
 
-  public HttpProxy(String username) {
-    super(username);
+  @Override
+  public int defaultPort() {
+    return DEFAULT_PORT;
   }
 
   /**
@@ -70,11 +74,12 @@ public HttpProxy(String username) {
   @Override
   public HttpProxy start(Vertx vertx) throws Exception {
     HttpServerOptions options = new HttpServerOptions();
-    options.setHost("localhost").setPort(PORT);
+    options.setHost("localhost").setPort(port);
     server = vertx.createHttpServer(options);
     server.requestHandler(request -> {
       HttpMethod method = request.method();
       String uri = request.uri();
+      String username = nextUserName();
       if (username != null) {
         String auth = request.getHeader("Proxy-Authorization");
         String expected = "Basic " + Base64.getEncoder().encodeToString((username + ":" + username).getBytes());
@@ -113,6 +118,7 @@ public HttpProxy start(Vertx vertx) throws Exception {
           NetClient netClient = vertx.createNetClient(netOptions);
           netClient.connect(port, host, ar1 -> {
             if (ar1.succeeded()) {
+              localAddresses.add(ar1.result().localAddress().toString());
               request.toNetSocket().onComplete(ar2 -> {
                 if (ar2.succeeded()) {
                   NetSocket serverSocket = ar2.result();
@@ -136,9 +142,16 @@ public HttpProxy start(Vertx vertx) throws Exception {
         if (forceUri != null) {
           uri = forceUri;
         }
-        HttpClient client = vertx.createHttpClient();
         RequestOptions opts = new RequestOptions();
         opts.setAbsoluteURI(uri);
+        HttpConnection serverConn = request.connection();
+        HttpClient client = clientMap.get(serverConn);
+        if (client == null) {
+          client = vertx.createHttpClient(new HttpClientOptions().setMaxPoolSize(1));
+          client.connectionHandler(conn -> localAddresses.add(conn.localAddress().toString()));
+          clientMap.put(serverConn, client);
+          serverConn.closeHandler(v -> clientMap.remove(serverConn));
+        }
         client.request(opts).compose(req -> {
           for (String name : request.headers().names()) {
             if (!name.equals("Proxy-Authorization")) {
@@ -200,11 +213,6 @@ public void stop() {
     }
   }
 
-  @Override
-  public int getPort() {
-    return PORT;
-  }
-
   @Override
   public HttpMethod getLastMethod() {
     return lastMethod;
diff --git a/src/test/java/io/vertx/test/proxy/Socks4Proxy.java b/src/test/java/io/vertx/test/proxy/Socks4Proxy.java
--- a/src/test/java/io/vertx/test/proxy/Socks4Proxy.java
+++ b/src/test/java/io/vertx/test/proxy/Socks4Proxy.java
@@ -37,7 +37,7 @@
  *
  * @author <a href="http://oss.lehmann.cx/">Alexander Lehmann</a>
  */
-public class Socks4Proxy extends TestProxyBase {
+public class Socks4Proxy extends TestProxyBase<Socks4Proxy> {
 
   private static final Logger log = LoggerFactory.getLogger(Socks4Proxy.class);
 
@@ -45,12 +45,13 @@ public class Socks4Proxy extends TestProxyBase {
   private static final Buffer connectResponse = Buffer.buffer(new byte[] { 0, 90, 0, 0, 0, 0, 0, 0 });
   private static final Buffer errorResponse = Buffer.buffer(new byte[] { 0, 91, 0, 0, 0, 0, 0, 0 });
 
-  private static final int PORT = 11080;
+  public static final int DEFAULT_PORT = 11080;
 
   private NetServer server;
 
-  public Socks4Proxy(String username) {
-    super(username);
+  @Override
+  public int defaultPort() {
+    return DEFAULT_PORT;
   }
 
   /**
@@ -62,7 +63,7 @@ public Socks4Proxy(String username) {
   @Override
   public Socks4Proxy start(Vertx vertx) throws Exception {
     NetServerOptions options = new NetServerOptions();
-    options.setHost("localhost").setPort(PORT);
+    options.setHost("localhost").setPort(port);
     server = vertx.createNetServer(options);
     server.connectHandler(socket -> {
       socket.handler(buffer -> {
@@ -75,7 +76,7 @@ public Socks4Proxy start(Vertx vertx) throws Exception {
         String ip = getByte4(buffer.getBuffer(4, 8));
 
         String authUsername = getString(buffer.getBuffer(8, buffer.length()));
-
+        String username = nextUserName();
         if (username != null && !authUsername.equals(username)) {
           log.debug("auth failed");
           log.debug("writing: " + toHex(errorResponse));
@@ -101,6 +102,7 @@ public Socks4Proxy start(Vertx vertx) throws Exception {
           NetClient netClient = vertx.createNetClient(new NetClientOptions());
           netClient.connect(port, host, result -> {
             if (result.succeeded()) {
+              localAddresses.add(result.result().localAddress().toString());
               log.debug("writing: " + toHex(connectResponse));
               socket.write(connectResponse);
               log.debug("connected, starting pump");
@@ -162,9 +164,4 @@ public void stop() {
       server = null;
     }
   }
-
-  @Override
-  public int getPort() {
-    return PORT;
-  }
 }
diff --git a/src/test/java/io/vertx/test/proxy/SocksProxy.java b/src/test/java/io/vertx/test/proxy/SocksProxy.java
--- a/src/test/java/io/vertx/test/proxy/SocksProxy.java
+++ b/src/test/java/io/vertx/test/proxy/SocksProxy.java
@@ -38,7 +38,7 @@
  *
  * @author <a href="http://oss.lehmann.cx/">Alexander Lehmann</a>
  */
-public class SocksProxy extends TestProxyBase {
+public class SocksProxy extends TestProxyBase<SocksProxy> {
 
   private static final Logger log = LoggerFactory.getLogger(SocksProxy.class);
 
@@ -53,12 +53,13 @@ public class SocksProxy extends TestProxyBase {
   private static final Buffer authSuccess = Buffer.buffer(new byte[] { 1, 0 });
   private static final Buffer authFailed = Buffer.buffer(new byte[] { 1, 1 });
 
-  private static final int PORT = 11080;
+  public static final int DEFAULT_PORT = 11080;
 
   private NetServer server;
 
-  public SocksProxy(String username) {
-    super(username);
+  @Override
+  public int defaultPort() {
+    return DEFAULT_PORT;
   }
 
   /**
@@ -70,10 +71,11 @@ public SocksProxy(String username) {
   @Override
   public SocksProxy start(Vertx vertx) throws Exception {
     NetServerOptions options = new NetServerOptions();
-    options.setHost("localhost").setPort(PORT);
+    options.setHost("localhost").setPort(port);
     server = vertx.createNetServer(options);
     server.connectHandler(socket -> {
       socket.handler(buffer -> {
+        String username = nextUserName();
         Buffer expectedInit = username == null ? clientInit : clientInitAuth;
         if (!buffer.equals(expectedInit)) {
           throw new IllegalStateException("expected " + toHex(expectedInit) + ", got " + toHex(buffer));
@@ -121,6 +123,7 @@ public SocksProxy start(Vertx vertx) throws Exception {
           NetClient netClient = vertx.createNetClient(new NetClientOptions());
           netClient.connect(port, host, result -> {
             if (result.succeeded()) {
+              localAddresses.add(result.result().localAddress().toString());
               log.debug("writing: " + toHex(connectResponse));
               socket.write(connectResponse);
               log.debug("connected, starting pump");
@@ -201,9 +204,4 @@ public void stop() {
       server = null;
     }
   }
-
-  @Override
-  public int getPort() {
-    return PORT;
-  }
 }
diff --git a/src/test/java/io/vertx/test/proxy/TestProxyBase.java b/src/test/java/io/vertx/test/proxy/TestProxyBase.java
--- a/src/test/java/io/vertx/test/proxy/TestProxyBase.java
+++ b/src/test/java/io/vertx/test/proxy/TestProxyBase.java
@@ -14,19 +14,69 @@
 import io.vertx.core.MultiMap;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpMethod;
+import io.vertx.core.net.SocketAddress;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Supplier;
 
 /**
  * @author <a href="http://oss.lehmann.cx/">Alexander Lehmann</a>
  *
  */
-public abstract class TestProxyBase {
+public abstract class TestProxyBase<P extends TestProxyBase<P>> {
 
-  protected final String username;
+  private Supplier<String> username;
+  protected int port;
   protected String lastUri;
   protected String forceUri;
+  protected List<String> localAddresses = Collections.synchronizedList(new ArrayList<>());
+
+  public TestProxyBase() {
+    port = defaultPort();
+  }
+
+  public P username(String username) {
+    this.username = () -> username;
+    return (P) this;
+  }
 
-  public TestProxyBase(String username) {
+  public P username(Supplier<String> username) {
     this.username = username;
+    return (P) this;
+  }
+
+  public P username(Collection<String> username) {
+    Iterator<String> it = username.iterator();
+    this.username = () -> it.hasNext() ? it.next() : null;
+    return (P) this;
+  }
+
+  public String nextUserName() {
+    return username != null ? username.get() : null;
+  }
+
+  public P port(int port) {
+    this.port = port;
+    return (P)this;
+  }
+
+  public int port() {
+    return port;
+  }
+
+  public abstract int defaultPort();
+
+  public String lastLocalAddress() {
+    int idx = localAddresses.size();
+    return idx == 0 ? null : localAddresses.get(idx - 1);
+  }
+
+  public List<String> localAddresses() {
+    return localAddresses;
   }
 
   /**
@@ -59,8 +109,6 @@ public MultiMap getLastRequestHeaders() {
     throw new UnsupportedOperationException();
   }
 
-  public abstract int getPort();
-
   public abstract TestProxyBase start(Vertx vertx) throws Exception;
   public abstract void stop();