diff --git a/src/main/java/com/ning/http/client/AsyncHttpClientConfig.java b/src/main/java/com/ning/http/client/AsyncHttpClientConfig.java index 593b0c4588..308cf074fa 100644 --- a/src/main/java/com/ning/http/client/AsyncHttpClientConfig.java +++ b/src/main/java/com/ning/http/client/AsyncHttpClientConfig.java @@ -79,6 +79,7 @@ public class AsyncHttpClientConfig { private final boolean allowSslConnectionPool; private final boolean useRawUrl; private final boolean removeQueryParamOnRedirect; + private final int ioThreadMultiplier; private final HostnameVerifier hostnameVerifier; private AsyncHttpClientConfig(int maxTotalConnections, @@ -106,6 +107,7 @@ private AsyncHttpClientConfig(int maxTotalConnections, boolean allowSslConnectionCaching, boolean useRawUrl, boolean removeQueryParamOnRedirect, + int ioThreadMultiplier, HostnameVerifier hostnameVerifier) { this.maxTotalConnections = maxTotalConnections; @@ -131,6 +133,7 @@ private AsyncHttpClientConfig(int maxTotalConnections, this.reaper = reaper; this.allowSslConnectionPool = allowSslConnectionCaching; this.removeQueryParamOnRedirect = removeQueryParamOnRedirect; + this.ioThreadMultiplier = ioThreadMultiplier; this.hostnameVerifier = hostnameVerifier; if (applicationThreadPool == null) { @@ -411,6 +414,14 @@ public boolean isRemoveQueryParamOnRedirect() { return removeQueryParamOnRedirect; } + /*** + * + * @return number to multiply by availableProcessors() that will determine # of NioWorkers to use + */ + public int getIoThreadMultiplier() { + return ioThreadMultiplier; + } + /** * Return true if one of the {@link java.util.concurrent.ExecutorService} has been shutdown. * @return true if one of the {@link java.util.concurrent.ExecutorService} has been shutdown. @@ -470,6 +481,7 @@ public Thread newThread(Runnable r) { private boolean allowSslConnectionPool = true; private boolean useRawUrl = false; private boolean removeQueryParamOnRedirect = true; + private int ioThreadMultiplier = 2; private HostnameVerifier hostnameVerifier = new HostnameVerifier() { public boolean verify( String s, SSLSession sslSession ) { @@ -865,6 +877,11 @@ public Builder setUseProxyProperties(boolean useProxyProperties) { return this; } + public Builder setIOThreadMultiplier(int multiplier){ + this.ioThreadMultiplier = multiplier; + return this; + } + /** * Set the {@link HostnameVerifier} * @param hostnameVerifier {@link HostnameVerifier} @@ -903,6 +920,7 @@ public Builder(AsyncHttpClientConfig prototype) { requestFilters.addAll(prototype.getRequestFilters()); responseFilters.addAll(prototype.getResponseFilters()); useRawUrl = prototype.isUseRawUrl(); + ioThreadMultiplier = prototype.getIoThreadMultiplier(); } /** @@ -946,6 +964,7 @@ public AsyncHttpClientConfig build() { allowSslConnectionPool, useRawUrl, removeQueryParamOnRedirect, + ioThreadMultiplier, hostnameVerifier); } } diff --git a/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java b/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java index 54eee3860f..fd422ba158 100644 --- a/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java +++ b/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java @@ -23,6 +23,7 @@ import com.ning.http.client.ConnectionsPool; import com.ning.http.client.Cookie; import com.ning.http.client.FluentCaseInsensitiveStringsMap; +import com.ning.http.client.FluentStringsMap; import com.ning.http.client.HttpResponseBodyPart; import com.ning.http.client.HttpResponseHeaders; import com.ning.http.client.HttpResponseStatus; @@ -194,7 +195,9 @@ public NettyAsyncHttpProvider(AsyncHttpClientConfig config) { } else { e = Executors.newCachedThreadPool(); } - socketChannelFactory = new NioClientSocketChannelFactory(e, config.executorService()); + int numWorkers = config.getIoThreadMultiplier()*Runtime.getRuntime().availableProcessors(); + log.info("numWorkers=" + numWorkers); + socketChannelFactory = new NioClientSocketChannelFactory(e, config.executorService(), numWorkers); } plainBootstrap = new ClientBootstrap(socketChannelFactory); secureBootstrap = new ClientBootstrap(socketChannelFactory); @@ -486,7 +489,8 @@ private static HttpRequest construct(AsyncHttpClientConfig config, HttpMethod m, URI uri, ChannelBuffer buffer) throws IOException { - String host = uri.getHost(); + + String host = AsyncHttpProviderUtils.getHost(uri); if (request.getVirtualHost() != null) { host = request.getVirtualHost(); @@ -880,7 +884,7 @@ private ListenableFuture doConnect(final Request request, final AsyncHand try { if (proxyServer == null || avoidProxy) { - channelFuture = bootstrap.connect(new InetSocketAddress(uri.getHost(), AsyncHttpProviderUtils.getPort(uri))); + channelFuture = bootstrap.connect(new InetSocketAddress(AsyncHttpProviderUtils.getHost(uri), AsyncHttpProviderUtils.getPort(uri))); } else { channelFuture = bootstrap.connect(new InetSocketAddress(proxyServer.getHost(), proxyServer.getPort())); } @@ -1282,7 +1286,7 @@ private Realm kerberosChallenge(List proxyAuth, NettyResponseFuture future) throws NTLMEngineException { URI uri = URI.create(request.getUrl()); - String host = request.getVirtualHost() == null ? uri.getHost() : request.getVirtualHost(); + String host = request.getVirtualHost() == null ? AsyncHttpProviderUtils.getHost(uri) : request.getVirtualHost(); String server = proxyServer == null ? host : proxyServer.getHost(); try { String challengeHeader = spnegoEngine.generateToken(server); diff --git a/src/main/java/com/ning/http/util/AsyncHttpProviderUtils.java b/src/main/java/com/ning/http/util/AsyncHttpProviderUtils.java index 7c2b3b74e2..a853ea4e30 100644 --- a/src/main/java/com/ning/http/util/AsyncHttpProviderUtils.java +++ b/src/main/java/com/ning/http/util/AsyncHttpProviderUtils.java @@ -175,6 +175,13 @@ public final static String getAuthority(URI uri) { return url; } + public final static String getHost(URI uri) { + String host = uri.getHost(); + if (host == null) { + host = uri.getAuthority(); + } + return host; + } public final static URI getRedirectUri(URI uri, String location) { URI newUri = uri.resolve(location); diff --git a/src/main/java/com/ning/http/util/ProxyUtils.java b/src/main/java/com/ning/http/util/ProxyUtils.java index c536530d96..0b126d6928 100644 --- a/src/main/java/com/ning/http/util/ProxyUtils.java +++ b/src/main/java/com/ning/http/util/ProxyUtils.java @@ -71,7 +71,7 @@ public class ProxyUtils { * @return true if we have to avoid proxy use (obeying non-proxy hosts settings), false otherwise. */ public static boolean avoidProxy(final ProxyServer proxyServer, final Request request) { - return avoidProxy(proxyServer, URI.create(request.getUrl()).getHost()); + return avoidProxy(proxyServer, AsyncHttpProviderUtils.getHost(URI.create(request.getUrl()))); } /**