fix(platform-vertx) Add LRUCache of HttpClient in ClientConnector.
* Add LRUCache of HttpClient in ClientConnector to make use of connection
pooling when connecting to similarly configured backends (e.g. same endpoint,   same security settings, etc).
* Clean up constructor by adding HttpConnectorOptions.
MODIFY
gateway/platforms/vertx3/vertx3/pom.xml
gateway/platforms/vertx3/vertx3/pom.xml
<groupId>io.netty</groupId>
<artifactId>netty-codec-http</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
<!-- Spec Only -->
<dependency>

MODIFY
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/connector/ConnectorFactory.java
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/connector/ConnectorFactory.java
@@ -25,11 +25,21 @@ import io.apiman.gateway.engine.auth.RequiredAuthType;
import io.apiman.gateway.engine.beans.Api;
import io.apiman.gateway.engine.beans.ApiRequest;
import io.apiman.gateway.engine.beans.exceptions.ConnectorException;
+import io.apiman.gateway.platforms.vertx3.http.HttpClientOptionsFactory;
import io.vertx.core.Vertx;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientOptions;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
/**
* Create Vert.x connectors to the enable apiman to connect to a backend API.
@@ -46,6 +56,24 @@ public class ConnectorFactory implements IConnectorFactory {
private Vertx vertx;
private TLSOptions tlsOptions;
+ private LoadingCache<HttpConnectorOptions, HttpClient> clientCache = CacheBuilder.newBuilder()
+ // TODO make this tuneable.
+ .maximumSize(500)
+ // Close any evicted connections.
+ .<HttpConnectorOptions, HttpClient>removalListener(eviction -> eviction.getValue().close())
+ // Either grab from cache or build new (which will be cached automatically).
+ .build(new CacheLoader<HttpConnectorOptions, HttpClient>() {
+
+ @Override
+ public HttpClient load(HttpConnectorOptions opts) throws Exception {
+ HttpClientOptions vxClientOptions = HttpClientOptionsFactory.parseTlsOptions(opts.getTlsOptions(), opts.getUri())
+ .setConnectTimeout(opts.getConnectionTimeout())
+ .setIdleTimeout(opts.getIdleTimeout())
+ .setKeepAlive(opts.isKeepAlive())
+ .setTryUseCompression(opts.isTryUseCompression());
+ return vertx.createHttpClient(vxClientOptions);
+ }
+ });
/**
* Constructor
@@ -57,17 +85,41 @@ public class ConnectorFactory implements IConnectorFactory {
this.tlsOptions = new TLSOptions(config);
}
+ // In the future we can switch to different back-end implementations here!
@Override
- public IApiConnector createConnector(ApiRequest request, final Api api, RequiredAuthType authType, boolean hasDataPolicy) {
+ public IApiConnector createConnector(ApiRequest request, Api api, RequiredAuthType authType, boolean hasDataPolicy) {
return new IApiConnector() {
@Override
public IApiConnection connect(ApiRequest request,
IAsyncResultHandler<IApiConnectionResponse> resultHandler)
throws ConnectorException {
- // In the future we can switch to different back-end implementations here!
- return new HttpConnector(vertx, api, request, authType, tlsOptions, hasDataPolicy, resultHandler);
+ HttpConnectorOptions httpOptions = new HttpConnectorOptions()
+ .setHasDataPolicy(hasDataPolicy)
+ .setRequiredAuthType(authType)
+ .setTlsOptions(tlsOptions)
+ .setUri(parseApiEndpoint(api));
+ // Get from cache
+ HttpClient client = clientFromCache(httpOptions);
+ return new HttpConnector(vertx, client, request, api, httpOptions, resultHandler);
}
};
}
+
+ private HttpClient clientFromCache(HttpConnectorOptions key) {
+ try {
+ return clientCache.get(key);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+
+ private URI parseApiEndpoint(Api api) {
+ try {
+ return new URI(api.getEndpoint());
+ } catch (URISyntaxException e) {
+ throw new RuntimeException(e);
+ }
+ }
}

MODIFY
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/connector/HttpConnector.java
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/connector/HttpConnector.java
package io.apiman.gateway.platforms.vertx3.connector;
import io.apiman.common.config.options.BasicAuthOptions;
-import io.apiman.common.config.options.TLSOptions;
import io.apiman.common.util.ApimanPathUtils;
import io.apiman.common.util.Basic;
import io.apiman.gateway.engine.IApiConnection;
@@ -34,7 +33,6 @@ import io.apiman.gateway.engine.io.IApimanBuffer;
import io.apiman.gateway.engine.io.ISignalReadStream;
import io.apiman.gateway.engine.io.ISignalWriteStream;
import io.apiman.gateway.platforms.vertx3.http.HttpApiFactory;
-import io.apiman.gateway.platforms.vertx3.http.HttpClientOptionsFactory;
import io.apiman.gateway.platforms.vertx3.i18n.Messages;
import io.apiman.gateway.platforms.vertx3.io.VertxApimanBuffer;
import io.vertx.core.Handler;
@@ -42,7 +40,6 @@ import io.vertx.core.MultiMap;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
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.HttpMethod;
@@ -50,8 +47,7 @@ import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import java.io.UnsupportedEncodingException;
-import java.net.MalformedURLException;
-import java.net.URL;
+import java.net.URI;
import java.net.URLEncoder;
import java.util.HashSet;
import java.util.Map.Entry;
@@ -96,48 +92,45 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
private String destination;
private int apiPort;
private boolean isHttps;
- private RequiredAuthType authType;
private BasicAuthOptions basicOptions;
private HttpClient client;
private HttpClientRequest clientRequest;
private HttpClientResponse clientResponse;
- private URL apiEndpoint;
+ private URI apiEndpoint;
+
+ private HttpConnectorOptions options;
- private boolean hasDataPolicies;
/**
* Construct an {@link HttpConnector} instance. The {@link #resultHandler} must remain exclusive to a
* given instance.
*
* @param vertx a vertx
- * @param api a api
+ * @param client the vertx http client
+ * @param api an API
* @param request a request with fields filled
- * @param authType the required auth type
- * @param tlsOptions the tls options
- * @param hasDataPolicy if the policy chain contains a data policy
+ * @param options the connector options
* @param resultHandler a handler, called when reading is permitted
*/
- public HttpConnector(Vertx vertx, Api api, ApiRequest request, RequiredAuthType authType,
- TLSOptions tlsOptions, boolean hasDataPolicy, IAsyncResultHandler<IApiConnectionResponse> resultHandler) {
+ public HttpConnector(Vertx vertx, HttpClient client, ApiRequest request, Api api, HttpConnectorOptions options,
+ IAsyncResultHandler<IApiConnectionResponse> resultHandler) {
+ this.client = client;
this.api = api;
this.apiRequest = request;
- this.authType = authType;
+
this.resultHandler = resultHandler;
this.exceptionHandler = new ExceptionHandler();
- this.hasDataPolicies = hasDataPolicy;
+ this.apiEndpoint = options.getUri();
+ this.options = options;
- apiEndpoint = parseApiEndpoint(api);
-
- isHttps = apiEndpoint.getProtocol().equals("https");
+ isHttps = apiEndpoint.getScheme().equals("https");
apiHost = apiEndpoint.getHost();
apiPort = getPort();
apiPath = apiEndpoint.getPath().isEmpty() || apiEndpoint.getPath().equals("/") ? "" : apiEndpoint.getPath();
destination = apiRequest.getDestination() == null ? "" : apiRequest.getDestination();
- HttpClientOptions clientOptions = HttpClientOptionsFactory.parseOptions(tlsOptions, apiEndpoint);
- this.client = vertx.createHttpClient(clientOptions);
verifyConnection();
doConnection();
}
@@ -150,7 +143,7 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
}
private void verifyConnection() {
- switch (authType) {
+ switch (options.getRequiredAuthType()) {
case BASIC:
basicOptions = new BasicAuthOptions(api.getEndpointProperties());
if (!isHttps && basicOptions.isRequireSSL())
@@ -199,7 +192,7 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
clientRequest.exceptionHandler(exceptionHandler);
- if (hasDataPolicies) {
+ if (options.hasDataPolicy()) {
clientRequest.headers().remove("Content-Length");
clientRequest.setChunked(true);
}
@@ -208,7 +201,7 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
addMandatoryRequestHeaders(clientRequest.headers());
- if (authType == RequiredAuthType.BASIC) {
+ if (options.getRequiredAuthType() == RequiredAuthType.BASIC) {
clientRequest.putHeader("Authorization", Basic.encode(basicOptions.getUsername(), basicOptions.getPassword()));
}
}
@@ -295,14 +288,6 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
return clientRequest.writeQueueFull();
}
- private URL parseApiEndpoint(Api api) {
- try {
- return new URL(api.getEndpoint());
- } catch (MalformedURLException e) {
- throw new RuntimeException(e);
- }
- }
-
private String queryParams(QueryMap queryParams) {
if (queryParams == null || queryParams.isEmpty())
return "";

ADD
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/connector/HttpConnectorOptions.java
+/*
+ * Copyright 2017 JBoss Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.apiman.gateway.platforms.vertx3.connector;
+
+import io.apiman.common.config.options.TLSOptions;
+import io.apiman.gateway.engine.auth.RequiredAuthType;
+import io.vertx.core.http.HttpClientOptions;
+
+import java.net.URI;
+
+/**
+ * Options for {@link HttpConnector}.
+ *
+ * @author Marc Savy {@literal <marc@rhymewithgravy.com>}
+ */
+public class HttpConnectorOptions {
+ private RequiredAuthType requiredAuthType = RequiredAuthType.DEFAULT;
+ private boolean hasDataPolicy = false;
+ private int connectionTimeout = HttpClientOptions.DEFAULT_CONNECT_TIMEOUT;
+ private int idleTimeout = HttpClientOptions.DEFAULT_IDLE_TIMEOUT;
+ private boolean keepAlive = HttpClientOptions.DEFAULT_KEEP_ALIVE;
+ private boolean tryUseCompression = HttpClientOptions.DEFAULT_TRY_USE_COMPRESSION;
+ private TLSOptions tlsOptions;
+ private URI endpoint;
+
+ /**
+ * @return the requiredAuthType
+ */
+ public RequiredAuthType getRequiredAuthType() {
+ return requiredAuthType;
+ }
+ /**
+ * @param requiredAuthType the requiredAuthType to set
+ * @return this
+ */
+ public HttpConnectorOptions setRequiredAuthType(RequiredAuthType requiredAuthType) {
+ this.requiredAuthType = requiredAuthType;
+ return this;
+ }
+ /**
+ * @return the hasDataPolicy
+ */
+ public boolean hasDataPolicy() {
+ return hasDataPolicy;
+ }
+ /**
+ * @param hasDataPolicy the hasDataPolicy to set
+ * @return this
+ */
+ public HttpConnectorOptions setHasDataPolicy(boolean hasDataPolicy) {
+ this.hasDataPolicy = hasDataPolicy;
+ return this;
+ }
+ /**
+ * @return the connectionTimeout
+ */
+ public int getConnectionTimeout() {
+ return connectionTimeout;
+ }
+ /**
+ * @param connectionTimeout the connectionTimeout to set
+ * @return this
+ */
+ public HttpConnectorOptions setConnectionTimeout(int connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ return this;
+ }
+ /**
+ * @return the idleTimeout
+ */
+ public int getIdleTimeout() {
+ return idleTimeout;
+ }
+ /**
+ * @param idleTimeout the idleTimeout to set
+ * @return this
+ */
+ public HttpConnectorOptions setIdleTimeout(int idleTimeout) {
+ this.idleTimeout = idleTimeout;
+ return this;
+ }
+ /**
+ * @return the keepAlive
+ */
+ public boolean isKeepAlive() {
+ return keepAlive;
+ }
+ /**
+ * @param keepAlive the keepAlive to set
+ * @return this
+ */
+ public HttpConnectorOptions setKeepAlive(boolean keepAlive) {
+ this.keepAlive = keepAlive;
+ return this;
+ }
+ /**
+ * @return the tryUseCompression
+ */
+ public boolean isTryUseCompression() {
+ return tryUseCompression;
+ }
+ /**
+ * @param tryUseCompression the tryUseCompression to set
+ * @return this
+ */
+ public HttpConnectorOptions setTryUseCompression(boolean tryUseCompression) {
+ this.tryUseCompression = tryUseCompression;
+ return this;
+ }
+
+ public TLSOptions getTlsOptions() {
+ return tlsOptions;
+ }
+
+ public HttpConnectorOptions setTlsOptions(TLSOptions tlsOptions) {
+ this.tlsOptions = tlsOptions;
+ return this;
+ }
+
+ public URI getUri() {
+ return endpoint;
+ }
+
+ public HttpConnectorOptions setUri(URI endpoint) {
+ this.endpoint = endpoint;
+ return this;
+ }
+
+ /* (non-Javadoc)
+ * @see java.lang.Object#hashCode()
+ */
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + connectionTimeout;
+ result = prime * result + idleTimeout;
+ result = prime * result + (keepAlive ? 1231 : 1237);
+ result = prime * result + ((tlsOptions == null) ? 0 : tlsOptions.hashCode());
+ result = prime * result + (tryUseCompression ? 1231 : 1237);
+ return result;
+ }
+ /* (non-Javadoc)
+ * @see java.lang.Object#equals(java.lang.Object)
+ */
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ HttpConnectorOptions other = (HttpConnectorOptions) obj;
+ if (connectionTimeout != other.connectionTimeout)
+ return false;
+ if (idleTimeout != other.idleTimeout)
+ return false;
+ if (keepAlive != other.keepAlive)
+ return false;
+ if (tlsOptions == null) {
+ if (other.tlsOptions != null)
+ return false;
+ } else if (!tlsOptions.equals(other.tlsOptions))
+ return false;
+ if (tryUseCompression != other.tryUseCompression)
+ return false;
+ return true;
+ }
+}

MODIFY
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/http/HttpClientOptionsFactory.java
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/http/HttpClientOptionsFactory.java
@@ -21,7 +21,7 @@ import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.core.net.JksOptions;
-import java.net.URL;
+import java.net.URI;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -41,7 +41,7 @@ public class HttpClientOptionsFactory {
private static Map<TLSOptions, HttpClientOptions> configCache = new HashMap<>();
private static Logger log = LoggerFactory.getLogger(HttpClientOptionsFactory.class);
- public static HttpClientOptions parseOptions(TLSOptions tlsOptions, URL apiEndpoint) {
+ public static HttpClientOptions parseTlsOptions(TLSOptions tlsOptions, URI apiEndpoint) {
if (configCache.containsKey(tlsOptions))
return configCache.get(tlsOptions);
@@ -50,10 +50,10 @@ public class HttpClientOptionsFactory {
return clientOptions;
}
- private static HttpClientOptions doParse(TLSOptions tlsOptions, URL apiEndpoint) {
+ private static HttpClientOptions doParse(TLSOptions tlsOptions, URI apiEndpoint) {
HttpClientOptions clientOptions = new HttpClientOptions();
- if (apiEndpoint.getProtocol().equals("http")) { //$NON-NLS-1$
+ if (apiEndpoint.getScheme().equals("http")) { //$NON-NLS-1$
return clientOptions.setSsl(false);
} else {
clientOptions.setSsl(true);

MODIFY
gateway/platforms/vertx3/vertx3/src/test/java/io/apiman/gateway/platforms/vertx3/connector/HttpConnectorDrainTest.java
gateway/platforms/vertx3/vertx3/src/test/java/io/apiman/gateway/platforms/vertx3/connector/HttpConnectorDrainTest.java
package io.apiman.gateway.platforms.vertx3.connector;
-import io.apiman.common.config.options.TLSOptions;
import io.apiman.gateway.engine.async.IAsyncHandler;
-import io.apiman.gateway.engine.auth.RequiredAuthType;
import io.apiman.gateway.engine.beans.Api;
import io.apiman.gateway.engine.beans.ApiRequest;
import io.apiman.gateway.engine.io.IApimanBuffer;
@@ -31,7 +29,7 @@ import io.vertx.ext.unit.Async;
import io.vertx.ext.unit.TestContext;
import io.vertx.ext.unit.junit.VertxUnitRunner;
-import java.util.Collections;
+import java.net.URI;
import org.junit.Assert;
import org.junit.Before;
@@ -104,7 +102,7 @@ public class HttpConnectorDrainTest {
}
@Test
- public void shouldTriggerDrainHandler(TestContext context) {
+ public void shouldTriggerDrainHandler(TestContext context) throws Exception {
Async asyncDrain = context.async(2);
Async asyncServer = context.async();
@@ -120,13 +118,15 @@ public class HttpConnectorDrainTest {
requestToPause.handler(data -> {});
}).listen(7297);
-
- HttpConnector httpConnector = new HttpConnector(Vertx.vertx(),
- api,
+ Vertx vertx = Vertx.vertx();
+ HttpConnector httpConnector = new HttpConnector(vertx,
+ vertx.createHttpClient(),
request,
- RequiredAuthType.DEFAULT,
- new TLSOptions(Collections.EMPTY_MAP),
- true, result -> {});
+ api,
+ new HttpConnectorOptions()
+ .setHasDataPolicy(true)
+ .setUri(URI.create(api.getEndpoint())),
+ result -> {});
// Should be fired when write queue reduces to acceptable size.
httpConnector.drainHandler(drain -> {

MODIFY
pom.xml
pom.xml
<version.commons-pool>1.6</version.commons-pool>
<version.com.zaxxer>2.4.0</version.com.zaxxer>
<version.com.google.code.gson>2.3.1</version.com.google.code.gson>
+ <version.com.google.guava>21.0</version.com.google.guava>
<version.io.searchbox.jest>0.1.7</version.io.searchbox.jest>
<version.io.prometheus>0.0.13</version.io.prometheus>
<version.javax.enterprise>1.2</version.javax.enterprise>
<artifactId>gson</artifactId>
<version>${version.com.google.code.gson}</version>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${version.com.google.guava}</version>
+ </dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>

fix(platform-vertx) Add LRUCache of HttpClient in ClientConnector.

* Add LRUCache of HttpClient in ClientConnector to make use of connection
  pooling when connecting to similarly configured backends (e.g. same endpoint,   same security settings, etc).

* Clean up constructor by adding HttpConnectorOptions.