Introduce OriginRequestProvider (#123)

Closes #35, closes #44

OriginRequestProvider is a Vert.x API type that unifies all the origin selection methods currently present as overloaded methods in HttpProxy.

OriginRequestProvider being a FunctionalInterface, users can provide one in the form of a lambda.

OriginRequestProvider is a little different from the existing contract.
Instead of the proxied request and HTTP client as parameters, it has the ProxyContext (that exposes all the proxy data and the client).

This allows implementations to make a decision on the modifications interceptors can make (including when the request is an upgrade to WebSocket). See #44

It also allows interceptor and provider to exchange data via the ProxyContext attachments. See #35

Signed-off-by: Thomas Segismont <[email protected]>

Author: tsegismont

src/main/asciidoc/index.adoc

@@ -48,16 +48,16 @@ All user-agent requests are forwarded to the *origin server* conveniently.
 
 === Origin server routing
 
-You can create a proxy that forwards all the traffic to a single server like seen before
+You can create a proxy that forwards all the traffic to a single server like seen before.
 
-You can set an origin selector to route the traffic to a given server
+You can set an origin selector to route the traffic to a given server:
 
 [source,java]
 ----
 {@link examples.HttpProxyExamples#originSelector}
 ----
 
-You can set a function to create the client request to the origin server for ultimate flexibility
+You can set a function to create the client request to the origin server for ultimate flexibility:
 
 [source,java]
 ----

src/main/java/examples/HttpProxyExamples.java

@@ -6,23 +6,17 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpClient;
 import io.vertx.core.http.HttpServer;
-import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.RequestOptions;
 import io.vertx.core.net.HostAndPort;
 import io.vertx.core.net.SocketAddress;
-import io.vertx.httpproxy.Body;
-import io.vertx.httpproxy.HttpProxy;
-import io.vertx.httpproxy.ProxyContext;
-import io.vertx.httpproxy.ProxyInterceptor;
-import io.vertx.httpproxy.ProxyOptions;
-import io.vertx.httpproxy.ProxyRequest;
-import io.vertx.httpproxy.ProxyResponse;
+import io.vertx.httpproxy.*;
 import io.vertx.httpproxy.cache.CacheOptions;
 
 /**
  * @author <a href="mailto:[email protected]">Emad Alblueshi</a>
  */
 
+@SuppressWarnings("unused")
 public class HttpProxyExamples {
 
   public void origin(Vertx vertx) {
@@ -46,20 +40,20 @@ public void proxy(Vertx vertx) {
     proxyServer.requestHandler(proxy).listen(8080);
   }
 
-  private SocketAddress resolveOriginAddress(HttpServerRequest request) {
+  private Future<SocketAddress> resolveOriginAddress(ProxyContext proxyContext) {
     return null;
   }
 
   public void originSelector(HttpProxy proxy) {
-    proxy.originSelector(request -> Future.succeededFuture(resolveOriginAddress(request)));
+    proxy.origin(OriginRequestProvider.selector(proxyContext -> resolveOriginAddress(proxyContext)));
   }
 
-  private RequestOptions resolveOriginOptions(HttpServerRequest request) {
+  private RequestOptions resolveOriginOptions(ProxyContext request) {
     return null;
   }
 
   public void originRequestProvider(HttpProxy proxy) {
-    proxy.originRequestProvider((request, client) -> client.request(resolveOriginOptions(request)));
+    proxy.origin((proxyContext) -> proxyContext.client().request(resolveOriginOptions(proxyContext)));
   }
 
   public void inboundInterceptor(HttpProxy proxy) {
@@ -137,34 +131,6 @@ private Body filter(Body body) {
     return body;
   }
 
-  public void more(Vertx vertx, HttpClient proxyClient) {
-    HttpProxy proxy = HttpProxy.reverseProxy(proxyClient).originSelector(
-      address -> Future.succeededFuture(SocketAddress.inetSocketAddress(7070, "origin"))
-    );
-  }
-
-  public void lowLevel(Vertx vertx, HttpServer proxyServer, HttpClient proxyClient) {
-
-    proxyServer.requestHandler(request -> {
-      ProxyRequest proxyRequest = ProxyRequest.reverseProxy(request);
-
-      proxyClient.request(proxyRequest.getMethod(), 8080, "origin", proxyRequest.getURI())
-        .compose(proxyRequest::send)
-        .onSuccess(proxyResponse -> {
-          // Send the proxy response
-          proxyResponse.send();
-        })
-        .onFailure(err -> {
-        // Release the request
-        proxyRequest.release();
-
-        // Send error
-        request.response().setStatusCode(500)
-          .send();
-      });
-    });
-  }
-
   public void overrideAuthority(HttpProxy proxy) {
     proxy.addInterceptor(new ProxyInterceptor() {
       @Override

src/main/java/io/vertx/httpproxy/HttpProxy.java

@@ -18,7 +18,6 @@
 import io.vertx.core.http.HttpClient;
 import io.vertx.core.http.HttpClientRequest;
 import io.vertx.core.http.HttpServerRequest;
-import io.vertx.core.http.RequestOptions;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.httpproxy.impl.ReverseProxy;
 
@@ -61,7 +60,7 @@ static HttpProxy reverseProxy(ProxyOptions options, HttpClient client) {
    */
   @Fluent
   default HttpProxy origin(SocketAddress address) {
-    return originSelector(req -> Future.succeededFuture(address));
+    return origin(OriginRequestProvider.fixedAddress(address));
   }
 
   /**
@@ -73,20 +72,20 @@ default HttpProxy origin(SocketAddress address) {
    */
   @Fluent
   default HttpProxy origin(int port, String host) {
-    return origin(SocketAddress.inetSocketAddress(port, host));
+    return origin(OriginRequestProvider.fixedAddress(port, host));
   }
 
   /**
    * Set a selector that resolves the <i><b>origin</b></i> address based on the incoming HTTP request.
    *
    * @param selector the selector
    * @return a reference to this, so the API can be used fluently
+   * @deprecated use {@link #origin(OriginRequestProvider)} instead
    */
+  @Deprecated
   @Fluent
   default HttpProxy originSelector(Function<HttpServerRequest, Future<SocketAddress>> selector) {
-    return originRequestProvider((req, client) -> selector
-      .apply(req)
-      .flatMap(server -> client.request(new RequestOptions().setServer(server))));
+    return origin(OriginRequestProvider.selector(proxyContext -> selector.apply(proxyContext.request().proxiedRequest())));
   }
 
   /**
@@ -95,10 +94,23 @@ default HttpProxy originSelector(Function<HttpServerRequest, Future<SocketAddres
    *
    * @param provider the provider
    * @return a reference to this, so the API can be used fluently
+   * @deprecated use {@link #origin(OriginRequestProvider)} instead
    */
-  @GenIgnore()
+  @Deprecated
+  @GenIgnore
   @Fluent
-  HttpProxy originRequestProvider(BiFunction<HttpServerRequest, HttpClient, Future<HttpClientRequest>> provider);
+  default HttpProxy originRequestProvider(BiFunction<HttpServerRequest, HttpClient, Future<HttpClientRequest>> provider) {
+    return origin(proxyContext -> provider.apply(proxyContext.request().proxiedRequest(), proxyContext.client()));
+  }
+
+  /**
+   * Set a provider that creates the request to the <i><b>origin</b></i> server based on {@link ProxyContext}.
+   *
+   * @param provider the provider
+   * @return a reference to this, so the API can be used fluently
+   */
+  @Fluent
+  HttpProxy origin(OriginRequestProvider provider);
 
   /**
    * Add an interceptor to the interceptor chain.

src/main/java/io/vertx/httpproxy/OriginRequestProvider.java

@@ -0,0 +1,56 @@
+package io.vertx.httpproxy;
+
+import io.vertx.codegen.annotations.VertxGen;
+import io.vertx.core.Future;
+import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.http.RequestOptions;
+import io.vertx.core.net.SocketAddress;
+
+import java.util.function.Function;
+
+/**
+ * A provider that creates the request to the <i><b>origin</b></i> server based on {@link ProxyContext}.
+ */
+@VertxGen
+@FunctionalInterface
+public interface OriginRequestProvider {
+
+  /**
+   * Creates a simple provider for a fixed {@code port} and {@code host}.
+   */
+  static OriginRequestProvider fixedAddress(int port, String host) {
+    return fixedAddress(SocketAddress.inetSocketAddress(port, host));
+  }
+
+  /**
+   * Creates a simple provider for a fixed {@link SocketAddress}.
+   */
+  static OriginRequestProvider fixedAddress(SocketAddress address) {
+    return new OriginRequestProvider() {
+      @Override
+      public Future<HttpClientRequest> create(ProxyContext proxyContext) {
+        return proxyContext.client().request(new RequestOptions().setServer(address));
+      }
+    };
+  }
+
+  /**
+   * Creates a provider that selects the <i><b>origin</b></i> server based on {@link ProxyContext}.
+   */
+  static OriginRequestProvider selector(Function<ProxyContext, Future<SocketAddress>> selector) {
+    return new OriginRequestProvider() {
+      @Override
+      public Future<HttpClientRequest> create(ProxyContext proxyContext) {
+        return selector.apply(proxyContext).flatMap(server -> proxyContext.client().request(new RequestOptions().setServer(server)));
+      }
+    };
+  }
+
+  /**
+   * Create the {@link HttpClientRequest} to the origin server for a given {@link ProxyContext}.
+   *
+   * @param proxyContext the context of the proxied request and response
+   * @return a future, completed with the {@link HttpClientRequest} or failed
+   */
+  Future<HttpClientRequest> create(ProxyContext proxyContext);
+}

src/main/java/io/vertx/httpproxy/ProxyContext.java

@@ -2,6 +2,7 @@
 
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.Future;
+import io.vertx.core.http.HttpClient;
 
 /**
  * A controller for proxy interception.
@@ -45,4 +46,9 @@ public interface ProxyContext {
    * @return the attached payload
    */
   <T> T get(String name, Class<T> type);
+
+  /**
+   * @return the {@link HttpClient} use to interact with the origin server
+   */
+  HttpClient client();
 }

src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java

@@ -20,7 +20,6 @@
 import io.vertx.httpproxy.spi.cache.Cache;
 
 import java.util.*;
-import java.util.function.BiFunction;
 
 import static io.vertx.core.http.HttpHeaders.CONNECTION;
 import static io.vertx.core.http.HttpHeaders.UPGRADE;
@@ -30,7 +29,7 @@ public class ReverseProxy implements HttpProxy {
   private final static Logger log = LoggerFactory.getLogger(ReverseProxy.class);
   private final HttpClient client;
   private final boolean supportWebSocket;
-  private BiFunction<HttpServerRequest, HttpClient, Future<HttpClientRequest>> selector = (req, client) -> Future.failedFuture("No origin available");
+  private OriginRequestProvider originRequestProvider = (pc) -> Future.failedFuture("No origin available");
   private final List<ProxyInterceptor> interceptors = new ArrayList<>();
 
   public ReverseProxy(ProxyOptions options, HttpClient client) {
@@ -44,8 +43,8 @@ public ReverseProxy(ProxyOptions options, HttpClient client) {
   }
 
   @Override
-  public HttpProxy originRequestProvider(BiFunction<HttpServerRequest, HttpClient, Future<HttpClientRequest>> provider) {
-    selector = provider;
+  public HttpProxy origin(OriginRequestProvider provider) {
+    originRequestProvider = Objects.requireNonNull(provider);
     return this;
   }
 
@@ -67,13 +66,14 @@ public void handle(HttpServerRequest request) {
       return;
     }
 
+    Proxy proxy = new Proxy(proxyRequest);
+
     // WebSocket upgrade tunneling
     if (supportWebSocket && io.vertx.core.http.impl.HttpUtils.canUpgradeToWebSocket(request)) {
-      handleWebSocketUpgrade(proxyRequest);
+      handleWebSocketUpgrade(proxy);
       return;
     }
 
-    Proxy proxy = new Proxy(proxyRequest);
     proxy.filters = interceptors.listIterator();
     proxy.sendRequest()
       .recover(throwable -> {
@@ -87,9 +87,10 @@ public void handle(HttpServerRequest request) {
       });
   }
 
-  private void handleWebSocketUpgrade(ProxyRequest proxyRequest) {
+  private void handleWebSocketUpgrade(ProxyContext proxyContext) {
+    ProxyRequest proxyRequest = proxyContext.request();
     HttpServerRequest proxiedRequest = proxyRequest.proxiedRequest();
-    resolveOrigin(proxiedRequest).onComplete(ar -> {
+    resolveOrigin(proxyContext).onComplete(ar -> {
       if (ar.succeeded()) {
         HttpClientRequest request = ar.result();
         request.setMethod(HttpMethod.GET);
@@ -148,8 +149,8 @@ private void end(ProxyRequest proxyRequest, int sc) {
       .send();
   }
 
-  private Future<HttpClientRequest> resolveOrigin(HttpServerRequest proxiedRequest) {
-    return selector.apply(proxiedRequest, client);
+  private Future<HttpClientRequest> resolveOrigin(ProxyContext proxyContext) {
+    return originRequestProvider.create(proxyContext);
   }
 
   private class Proxy implements ProxyContext {
@@ -174,6 +175,11 @@ public <T> T get(String name, Class<T> type) {
       return type.isInstance(o) ? type.cast(o) : null;
     }
 
+    @Override
+    public HttpClient client() {
+      return client;
+    }
+
     @Override
     public ProxyRequest request() {
       return request;
@@ -205,7 +211,7 @@ public Future<Void> sendResponse() {
     }
 
     private Future<ProxyResponse> sendProxyRequest(ProxyRequest proxyRequest) {
-      return resolveOrigin(proxyRequest.proxiedRequest()).compose(proxyRequest::send);
+      return resolveOrigin(this).compose(proxyRequest::send);
     }
 
     private Future<Void> sendProxyResponse(ProxyResponse response) {

src/test/java/io/vertx/httpproxy/ProxyTest.java

@@ -14,6 +14,7 @@
 import io.vertx.core.http.HttpClient;
 import io.vertx.core.http.HttpClientResponse;
 import io.vertx.core.http.HttpMethod;
+import io.vertx.core.http.RequestOptions;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
@@ -49,7 +50,7 @@ public void testRoundRobinSelector(TestContext ctx) {
       backends[i] = startHttpBackend(ctx, 8081 + value, req -> req.response().end("" + value));
     }
     AtomicInteger count = new AtomicInteger();
-    startProxy(proxy -> proxy.originSelector(req -> Future.succeededFuture(backends[count.getAndIncrement() % backends.length])));
+    startProxy(proxy -> proxy.origin(OriginRequestProvider.selector(proxyContext -> Future.succeededFuture(backends[count.getAndIncrement() % backends.length]))));
     HttpClient client = vertx.createHttpClient();
     Map<String, AtomicInteger> result = Collections.synchronizedMap(new HashMap<>());
     Async latch = ctx.async();
@@ -196,4 +197,29 @@ public Future<Void> handleProxyResponse(ProxyContext context) {
       }))
       .onComplete(ctx.asyncAssertSuccess(body -> async.complete()));
   }
+
+  @Test
+  public void testVariableFromInterceptor(TestContext ctx) {
+    SocketAddress backend = startHttpBackend(ctx, 8081, req -> req.response().end("HOLA"));
+    ProxyInterceptor interceptor = new ProxyInterceptor() {
+      @Override
+      public Future<ProxyResponse> handleProxyRequest(ProxyContext context) {
+        context.set("foo", "bar");
+        return context.sendRequest();
+      }
+    };
+    OriginRequestProvider provider = (proxyContext) -> {
+      ctx.assertEquals("bar", proxyContext.get("foo", String.class));
+      return proxyContext.client().request(new RequestOptions().setServer(backend));
+    };
+    startProxy(proxy -> proxy.origin(provider).addInterceptor(interceptor));
+    HttpClient client = vertx.createHttpClient();
+    client
+      .request(HttpMethod.GET, 8080, "localhost", "/")
+      .compose(req -> req
+        .send()
+        .compose(HttpClientResponse::body)
+      )
+      .onComplete(ctx.asyncAssertSuccess(buffer -> ctx.assertEquals("HOLA", buffer.toString())));
+  }
 }