Skip to content

Commit 1b986fd

Browse files
author
Stephane Landelle
committed
Proper fix for AsyncHttpClient#115 in 1.7.x branch
1 parent b6a5cb4 commit 1b986fd

File tree

4 files changed

+59
-67
lines changed

4 files changed

+59
-67
lines changed

‎src/main/java/com/ning/http/client/ProxyServer.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ public String getPassword() {
102102
return password;
103103
}
104104

105-
public URI getUri() {
105+
public URI getURI() {
106106
return uri;
107107
}
108108

‎src/main/java/com/ning/http/client/providers/grizzly/GrizzlyAsyncHttpProvider.java

Lines changed: 8 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -2283,7 +2283,7 @@ void doAsyncTrackedConnection(final Request request,
22832283
final GrizzlyResponseFuture requestFuture,
22842284
final CompletionHandler<Connection> connectHandler)
22852285
throws IOException, ExecutionException, InterruptedException {
2286-
Connection c = pool.poll(getPoolKey(request));
2286+
Connection c = pool.poll(getPoolKey(request, requestFuture.getProxy()));
22872287
if (c == null) {
22882288
if (!connectionMonitor.acquire()) {
22892289
throw new IOException("Max connections exceeded");
@@ -2331,8 +2331,8 @@ private Connection obtainConnection0(final Request request,
23312331

23322332
final URI uri = request.getURI();
23332333
final ProxyServer proxy = requestFuture.getProxy();
2334-
String host = ((proxy != null) ? proxy.getHost() : uri.getHost());
2335-
int port = ((proxy != null) ? proxy.getPort() : uri.getPort());
2334+
String host = (proxy != null) ? proxy.getHost() : uri.getHost();
2335+
int port = (proxy != null) ? proxy.getPort() : uri.getPort();
23362336
int cTimeout = provider.clientConfig.getConnectionTimeoutInMs();
23372337
FutureImpl<Connection> future = Futures.createSafeFuture();
23382338
CompletionHandler<Connection> ch = Futures.toCompletionHandler(future,
@@ -2348,26 +2348,16 @@ private Connection obtainConnection0(final Request request,
23482348
}
23492349
}
23502350

2351-
private ProxyServer getProxyServer(Request request) {
2352-
2353-
ProxyServer proxyServer = request.getProxyServer();
2354-
if (proxyServer == null) {
2355-
proxyServer = provider.clientConfig.getProxyServer();
2356-
}
2357-
return proxyServer;
2358-
2359-
}
2360-
23612351
boolean returnConnection(final Request request, final Connection c) {
2352+
ProxyServer proxyServer = ProxyUtils.getProxyServer(provider.clientConfig, request);
23622353
final boolean result = (DO_NOT_CACHE.get(c) == null
2363-
&& pool.offer(getPoolKey(request), c));
2354+
&& pool.offer(getPoolKey(request, proxyServer), c));
23642355
if (result) {
23652356
if (provider.resolver != null) {
23662357
provider.resolver.setTimeoutMillis(c, IdleTimeoutFilter.FOREVER);
23672358
}
23682359
}
23692360
return result;
2370-
23712361
}
23722362

23732363

@@ -2421,9 +2411,9 @@ public void updated(Connection result) {
24212411
};
24222412
}
24232413

2424-
private static String getPoolKey(final Request request) {
2425-
final ConnectionPoolKeyStrategy keyStrategy = request.getConnectionPoolKeyStrategy();
2426-
return keyStrategy.getKey(request.getURI());
2414+
private static String getPoolKey(Request request, ProxyServer proxyServer) {
2415+
URI uri = proxyServer != null? proxyServer.getURI(): request.getURI();
2416+
return request.getConnectionPoolKeyStrategy().getKey(uri);
24272417
}
24282418

24292419
// ------------------------------------------------------ Nested Classes

‎src/main/java/com/ning/http/client/providers/jdk/JDKAsyncHttpProvider.java

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -163,11 +163,10 @@ public <T> ListenableFuture<T> execute(Request request, AsyncHandler<T> handler,
163163
}
164164

165165
private HttpURLConnection createUrlConnection(Request request) throws IOException {
166-
ProxyServer proxyServer = request.getProxyServer() != null ? request.getProxyServer() : config.getProxyServer();
166+
ProxyServer proxyServer = ProxyUtils.getProxyServer(config, request);
167167
Realm realm = request.getRealm() != null ? request.getRealm() : config.getRealm();
168-
boolean avoidProxy = ProxyUtils.avoidProxy(proxyServer, request);
169168
Proxy proxy = null;
170-
if (!avoidProxy && proxyServer != null || realm != null) {
169+
if (proxyServer != null || realm != null) {
171170
try {
172171
proxy = configureProxyAndAuth(proxyServer, realm);
173172
} catch (AuthenticationException e) {
@@ -177,10 +176,9 @@ private HttpURLConnection createUrlConnection(Request request) throws IOExceptio
177176

178177
HttpURLConnection urlConnection = null;
179178
if (proxy == null) {
180-
urlConnection =
181-
(HttpURLConnection) AsyncHttpProviderUtils.createUri(request.getUrl()).toURL().openConnection(Proxy.NO_PROXY);
179+
urlConnection = (HttpURLConnection) request.getURI().toURL().openConnection(Proxy.NO_PROXY);
182180
} else {
183-
urlConnection = (HttpURLConnection) AsyncHttpProviderUtils.createUri(request.getUrl()).toURL().openConnection(proxy);
181+
urlConnection = (HttpURLConnection) proxyServer.getURI().toURL().openConnection(proxy);
184182
}
185183

186184
if (request.getUrl().startsWith("https")) {

‎src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

Lines changed: 46 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -201,25 +201,25 @@ public NettyAsyncHttpProvider(AsyncHttpClientConfig config) {
201201
socketChannelFactory = new OioClientSocketChannelFactory(config.executorService());
202202
this.allowReleaseSocketChannelFactory = true;
203203
} else {
204-
// check if external NioClientSocketChannelFactory is defined
204+
// check if external NioClientSocketChannelFactory is defined
205205
Object oo = asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.SOCKET_CHANNEL_FACTORY);
206206
if (oo != null && NioClientSocketChannelFactory.class.isAssignableFrom(oo.getClass())) {
207-
this.socketChannelFactory = NioClientSocketChannelFactory.class.cast(oo);
207+
this.socketChannelFactory = NioClientSocketChannelFactory.class.cast(oo);
208208

209-
// cannot allow releasing shared channel factory
210-
this.allowReleaseSocketChannelFactory = false;
209+
// cannot allow releasing shared channel factory
210+
this.allowReleaseSocketChannelFactory = false;
211211
} else {
212-
ExecutorService e;
213-
Object o = asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.BOSS_EXECUTOR_SERVICE);
214-
if (o != null && ExecutorService.class.isAssignableFrom(o.getClass())) {
215-
e = ExecutorService.class.cast(o);
216-
} else {
217-
e = Executors.newCachedThreadPool();
218-
}
219-
int numWorkers = config.getIoThreadMultiplier() * Runtime.getRuntime().availableProcessors();
220-
log.debug("Number of application's worker threads is {}", numWorkers);
221-
socketChannelFactory = new NioClientSocketChannelFactory(e, config.executorService(), numWorkers);
222-
this.allowReleaseSocketChannelFactory = true;
212+
ExecutorService e;
213+
Object o = asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.BOSS_EXECUTOR_SERVICE);
214+
if (o != null && ExecutorService.class.isAssignableFrom(o.getClass())) {
215+
e = ExecutorService.class.cast(o);
216+
} else {
217+
e = Executors.newCachedThreadPool();
218+
}
219+
int numWorkers = config.getIoThreadMultiplier() * Runtime.getRuntime().availableProcessors();
220+
log.debug("Number of application's worker threads is {}", numWorkers);
221+
socketChannelFactory = new NioClientSocketChannelFactory(e, config.executorService(), numWorkers);
222+
this.allowReleaseSocketChannelFactory = true;
223223
}
224224
}
225225
plainBootstrap = new ClientBootstrap(socketChannelFactory);
@@ -629,7 +629,7 @@ private static HttpRequest construct(AsyncHttpClientConfig config,
629629
nettyRequest = new DefaultHttpRequest(HttpVersion.HTTP_1_0, m, AsyncHttpProviderUtils.getAuthority(uri));
630630
} else {
631631
String path = null;
632-
if (proxyServer != null && !isSecure(uri) && !config.isUseRelativeURIsWithSSLProxies())
632+
if (proxyServer != null && !(isSecure(uri) && config.isUseRelativeURIsWithSSLProxies()))
633633
path = uri.toString();
634634
else if (uri.getRawQuery() != null)
635635
path = uri.getRawPath() + "?" + uri.getRawQuery();
@@ -901,11 +901,11 @@ public void close() {
901901
config.executorService().shutdown();
902902
config.reaper().shutdown();
903903
if (this.allowReleaseSocketChannelFactory) {
904-
socketChannelFactory.releaseExternalResources();
905-
plainBootstrap.releaseExternalResources();
906-
secureBootstrap.releaseExternalResources();
907-
webSocketBootstrap.releaseExternalResources();
908-
secureWebSocketBootstrap.releaseExternalResources();
904+
socketChannelFactory.releaseExternalResources();
905+
plainBootstrap.releaseExternalResources();
906+
secureBootstrap.releaseExternalResources();
907+
webSocketBootstrap.releaseExternalResources();
908+
secureWebSocketBootstrap.releaseExternalResources();
909909
}
910910
} catch (Throwable t) {
911911
log.warn("Unexpected error on close", t);
@@ -956,7 +956,8 @@ private <T> ListenableFuture<T> doConnect(final Request request, final AsyncHand
956956
if (f != null && f.reuseChannel() && f.channel() != null) {
957957
channel = f.channel();
958958
} else {
959-
channel = lookupInCache(uri, request.getConnectionPoolKeyStrategy());
959+
URI connectionKeyUri = useProxy? proxyServer.getURI() : uri;
960+
channel = lookupInCache(connectionKeyUri, request.getConnectionPoolKeyStrategy());
960961
}
961962
}
962963

@@ -1306,18 +1307,24 @@ private Realm ntlmProxyChallenge(List<String> wwwAuth,
13061307
} else {
13071308
realmBuilder = new Realm.RealmBuilder();
13081309
}
1309-
newRealm = realmBuilder//.setScheme(realm.getAuthScheme())
1310+
newRealm = realmBuilder
13101311
.setUri(request.getURI().getPath())
13111312
.setMethodName(request.getMethod())
13121313
.build();
13131314

13141315
return newRealm;
13151316
}
1317+
1318+
private String getPoolKey(NettyResponseFuture<?> future) throws MalformedURLException {
1319+
URI uri = future.getProxyServer() != null ? future.getProxyServer().getURI() : future.getURI();
1320+
return future.getConnectionPoolKeyStrategy().getKey(uri);
1321+
}
13161322

1317-
private void drainChannel(final ChannelHandlerContext ctx, final NettyResponseFuture<?> future, final boolean keepAlive, final URI uri) {
1323+
private void drainChannel(final ChannelHandlerContext ctx, final NettyResponseFuture<?> future) {
13181324
ctx.setAttachment(new AsyncCallable(future) {
13191325
public Object call() throws Exception {
1320-
if (keepAlive && ctx.getChannel().isReadable() && connectionsPool.offer(future.getConnectionPoolKeyStrategy().getKey(uri), ctx.getChannel())) {
1326+
1327+
if (future.getKeepAlive() && ctx.getChannel().isReadable() && connectionsPool.offer(getPoolKey(future), ctx.getChannel())) {
13211328
return null;
13221329
}
13231330

@@ -1353,7 +1360,7 @@ private void replayRequest(final NettyResponseFuture<?> future, FilterContext fc
13531360
future.touch();
13541361

13551362
log.debug("\n\nReplaying Request {}\n for Future {}\n", newRequest, future);
1356-
drainChannel(ctx, future, future.getKeepAlive(), future.getURI());
1363+
drainChannel(ctx, future);
13571364
nextRequest(newRequest, future);
13581365
return;
13591366
}
@@ -1510,10 +1517,9 @@ private void markAsDone(final NettyResponseFuture<?> future, final ChannelHandle
15101517

15111518
private void finishUpdate(final NettyResponseFuture<?> future, final ChannelHandlerContext ctx, boolean lastValidChunk) throws IOException {
15121519
if (lastValidChunk && future.getKeepAlive()) {
1513-
drainChannel(ctx, future, future.getKeepAlive(), future.getURI());
1520+
drainChannel(ctx, future);
15141521
} else {
1515-
if (future.getKeepAlive() && ctx.getChannel().isReadable() &&
1516-
connectionsPool.offer(future.getConnectionPoolKeyStrategy().getKey(future.getURI()), ctx.getChannel())) {
1522+
if (future.getKeepAlive() && ctx.getChannel().isReadable() && connectionsPool.offer(getPoolKey(future), ctx.getChannel())) {
15171523
markAsDone(future, ctx);
15181524
return;
15191525
}
@@ -2066,8 +2072,8 @@ private boolean redirect(Request request,
20662072
&& config.isStrict302Handling())) {
20672073
nBuilder.setMethod("GET");
20682074
}
2069-
final URI initialConnectionUri = future.getURI();
20702075
final boolean initialConnectionKeepAlive = future.getKeepAlive();
2076+
final String initialPoolKey = getPoolKey(future);
20712077
future.setURI(uri);
20722078
String newUrl = uri.toString();
20732079
if (request.getUrl().startsWith(WEBSOCKET)) {
@@ -2085,11 +2091,9 @@ private boolean redirect(Request request,
20852091
nBuilder.addOrReplaceCookie(c);
20862092
}
20872093

2088-
final String connectionPoolKey = future.getConnectionPoolKeyStrategy().getKey(initialConnectionUri);
20892094
AsyncCallable ac = new AsyncCallable(future) {
20902095
public Object call() throws Exception {
2091-
if (initialConnectionKeepAlive && ctx.getChannel().isReadable() &&
2092-
connectionsPool.offer(connectionPoolKey, ctx.getChannel())) {
2096+
if (initialConnectionKeepAlive && ctx.getChannel().isReadable() && connectionsPool.offer(initialPoolKey, ctx.getChannel())) {
20932097
return null;
20942098
}
20952099
finishChannel(ctx);
@@ -2186,7 +2190,7 @@ public void handle(final ChannelHandlerContext ctx, final MessageEvent e) throws
21862190
//}
21872191

21882192
if (statusCode == 401
2189-
&& realm != null
2193+
&& realm != null
21902194
&& wwwAuth.size() > 0
21912195
&& !future.getAndSetAuth(true)) {
21922196

@@ -2220,7 +2224,7 @@ public void handle(final ChannelHandlerContext ctx, final MessageEvent e) throws
22202224
log.debug("Sending authentication to {}", request.getUrl());
22212225
AsyncCallable ac = new AsyncCallable(future) {
22222226
public Object call() throws Exception {
2223-
drainChannel(ctx, future, future.getKeepAlive(), future.getURI());
2227+
drainChannel(ctx, future);
22242228
nextRequest(builder.setHeaders(headers).setRealm(nr).build(), future);
22252229
return null;
22262230
}
@@ -2244,7 +2248,7 @@ public Object call() throws Exception {
22442248

22452249
List<String> proxyAuth = getAuthorizationToken(response.getHeaders(), HttpHeaders.Names.PROXY_AUTHENTICATE);
22462250
if (statusCode == 407
2247-
&& realm != null
2251+
&& realm != null
22482252
&& proxyAuth.size() > 0
22492253
&& !future.getAndSetAuth(true)) {
22502254

@@ -2310,7 +2314,7 @@ public Object call() throws Exception {
23102314
if (nettyRequest.getMethod().equals(HttpMethod.HEAD)) {
23112315
updateBodyAndInterrupt(future, handler, new ResponseBodyPart(future.getURI(), response, NettyAsyncHttpProvider.this, true));
23122316
markAsDone(future, ctx);
2313-
drainChannel(ctx, future, future.getKeepAlive(), future.getURI());
2317+
drainChannel(ctx, future);
23142318
}
23152319

23162320
} else if (e.getMessage() instanceof HttpChunk) {
@@ -2363,9 +2367,9 @@ private final class WebSocketProtocol implements Protocol {
23632367
private static final byte OPCODE_BINARY = 0x2;
23642368
private static final byte OPCODE_UNKNOWN = -1;
23652369

2366-
protected ChannelBuffer byteBuffer = null;
2367-
protected StringBuilder textBuffer = null;
2368-
protected byte pendingOpcode = OPCODE_UNKNOWN;
2370+
protected ChannelBuffer byteBuffer = null;
2371+
protected StringBuilder textBuffer = null;
2372+
protected byte pendingOpcode = OPCODE_UNKNOWN;
23692373

23702374
// @Override
23712375
public void handle(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
@@ -2444,10 +2448,10 @@ public void handle(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
24442448
final WebSocketFrame frame = (WebSocketFrame) e.getMessage();
24452449

24462450
if(frame instanceof TextWebSocketFrame) {
2447-
pendingOpcode = OPCODE_TEXT;
2451+
pendingOpcode = OPCODE_TEXT;
24482452
}
24492453
else if(frame instanceof BinaryWebSocketFrame) {
2450-
pendingOpcode = OPCODE_BINARY;
2454+
pendingOpcode = OPCODE_BINARY;
24512455
}
24522456

24532457
HttpChunk webSocketChunk = new HttpChunk() {

0 commit comments

Comments
 (0)