Skip to content

Commit b2ff86b

Browse files
jfarcandjfarcand
committed
Merge pull request AsyncHttpClient#23 from rlubke/master
Several fixes for the Grizzly provider.
2 parents d4e7833 + 3407e5c commit b2ff86b

File tree

4 files changed

+164
-96
lines changed

4 files changed

+164
-96
lines changed

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

Lines changed: 96 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@
9696
import java.io.UnsupportedEncodingException;
9797
import java.net.InetSocketAddress;
9898
import java.net.URI;
99+
import java.net.URL;
99100
import java.net.URLEncoder;
100101
import java.nio.ByteBuffer;
101102
import java.security.NoSuchAlgorithmException;
@@ -401,7 +402,7 @@ public void updated(WriteResult result) {
401402
}
402403

403404

404-
private void setHttpTransactionContext(final AttributeStorage storage,
405+
void setHttpTransactionContext(final AttributeStorage storage,
405406
final HttpTransactionContext httpTransactionState) {
406407

407408
if (httpTransactionState == null) {
@@ -412,22 +413,22 @@ private void setHttpTransactionContext(final AttributeStorage storage,
412413

413414
}
414415

415-
private HttpTransactionContext getHttpTransactionContext(final AttributeStorage storage) {
416+
HttpTransactionContext getHttpTransactionContext(final AttributeStorage storage) {
416417

417418
return REQUEST_STATE_ATTR.get(storage);
418419

419420
}
420421

421422

422-
private void timeout(final Connection c) {
423+
void timeout(final Connection c) {
423424

424425
final HttpTransactionContext context = getHttpTransactionContext(c);
425426
setHttpTransactionContext(c, null);
426427
context.abort(new TimeoutException("Timeout exceeded"));
427428

428429
}
429430

430-
private static int getPort(final URI uri, final int p) {
431+
static int getPort(final URI uri, final int p) {
431432
int port = p;
432433
if (port == -1) {
433434
final String protocol = uri.getScheme().toLowerCase();
@@ -444,9 +445,9 @@ private static int getPort(final URI uri, final int p) {
444445

445446

446447
@SuppressWarnings({"unchecked"})
447-
private void sendRequest(final FilterChainContext ctx,
448-
final Request request,
449-
final HttpRequestPacket requestPacket)
448+
void sendRequest(final FilterChainContext ctx,
449+
final Request request,
450+
final HttpRequestPacket requestPacket)
450451
throws IOException {
451452

452453
if (requestHasEntityBody(request)) {
@@ -497,26 +498,27 @@ boolean handleStatus(final HttpResponsePacket httpResponse,
497498
} // END StatusHandler
498499

499500

500-
private final class HttpTransactionContext {
501+
final class HttpTransactionContext {
501502

502-
private final AtomicInteger redirectCount = new AtomicInteger(0);
503+
final AtomicInteger redirectCount = new AtomicInteger(0);
503504

504-
private final int maxRedirectCount;
505-
private final boolean redirectsAllowed;
506-
private final GrizzlyAsyncHttpProvider provider =
505+
final int maxRedirectCount;
506+
final boolean redirectsAllowed;
507+
final GrizzlyAsyncHttpProvider provider =
507508
GrizzlyAsyncHttpProvider.this;
508509

509-
private Request request;
510-
private AsyncHandler handler;
511-
private BodyHandler bodyHandler;
512-
private StatusHandler statusHandler;
513-
private StatusHandler.InvocationStatus invocationStatus =
510+
Request request;
511+
String requestUrl;
512+
AsyncHandler handler;
513+
BodyHandler bodyHandler;
514+
StatusHandler statusHandler;
515+
StatusHandler.InvocationStatus invocationStatus =
514516
StatusHandler.InvocationStatus.CONTINUE;
515-
private GrizzlyResponseStatus responseStatus;
516-
private GrizzlyResponseFuture future;
517-
private String lastRedirectURI;
518-
private AtomicLong totalBodyWritten = new AtomicLong();
519-
private AsyncHandler.STATE currentState;
517+
GrizzlyResponseStatus responseStatus;
518+
GrizzlyResponseFuture future;
519+
String lastRedirectURI;
520+
AtomicLong totalBodyWritten = new AtomicLong();
521+
AsyncHandler.STATE currentState;
520522

521523

522524
// -------------------------------------------------------- Constructors
@@ -531,14 +533,15 @@ private final class HttpTransactionContext {
531533
this.handler = handler;
532534
redirectsAllowed = provider.clientConfig.isRedirectEnabled();
533535
maxRedirectCount = provider.clientConfig.getMaxRedirects();
536+
this.requestUrl = request.getUrl();
534537

535538
}
536539

537540

538541
// ----------------------------------------------------- Private Methods
539542

540543

541-
private HttpTransactionContext copy() {
544+
HttpTransactionContext copy() {
542545
final HttpTransactionContext newContext =
543546
new HttpTransactionContext(future,
544547
request,
@@ -554,20 +557,20 @@ private HttpTransactionContext copy() {
554557
}
555558

556559

557-
private void abort(final Throwable t) {
560+
void abort(final Throwable t) {
558561
if (future != null) {
559562
future.abort(t);
560563
}
561564
}
562565

563-
private void done(final Callable c) {
566+
void done(final Callable c) {
564567
if (future != null) {
565568
future.done(c);
566569
}
567570
}
568571

569572
@SuppressWarnings({"unchecked"})
570-
private void result(Object result) {
573+
void result(Object result) {
571574
if (future != null) {
572575
future.delegate.result(result);
573576
future.done(null);
@@ -699,7 +702,8 @@ private void sendAsGrizzlyRequest(final Request request,
699702
final FilterChainContext ctx)
700703
throws IOException {
701704

702-
final URI uri = AsyncHttpProviderUtils.createUri(request.getUrl());
705+
final HttpTransactionContext httpCtx = getHttpTransactionContext(ctx.getConnection());
706+
final URI uri = AsyncHttpProviderUtils.createUri(httpCtx.requestUrl);
703707
final HttpRequestPacket.Builder builder = HttpRequestPacket.builder();
704708

705709
builder.method(request.getMethod());
@@ -756,7 +760,7 @@ private void sendAsGrizzlyRequest(final Request request,
756760
}
757761
}
758762
}
759-
final AsyncHandler h = getHttpTransactionContext(ctx.getConnection()).handler;
763+
final AsyncHandler h = httpCtx.handler;
760764
if (TransferCompletionHandler.class.isAssignableFrom(h.getClass())) {
761765
final FluentCaseInsensitiveStringsMap map =
762766
new FluentCaseInsensitiveStringsMap(request.getHeaders());
@@ -954,10 +958,10 @@ protected void onHttpContentEncoded(HttpContent content, FilterChainContext ctx)
954958
final AsyncHandler handler = context.handler;
955959
if (TransferCompletionHandler.class.isAssignableFrom(handler.getClass())) {
956960
final int written = content.getContent().remaining();
957-
context.totalBodyWritten.addAndGet(written);
961+
final long total = context.totalBodyWritten.addAndGet(written);
958962
((TransferCompletionHandler) handler).onContentWriteProgress(
959963
written,
960-
context.totalBodyWritten.get(),
964+
total,
961965
content.getHttpHeader().getContentLength());
962966
}
963967
}
@@ -1012,7 +1016,7 @@ protected void onInitialLineParsed(HttpHeader httpHeader,
10121016
}
10131017
final GrizzlyResponseStatus responseStatus =
10141018
new GrizzlyResponseStatus((HttpResponsePacket) httpHeader,
1015-
getURI(context.request.getUrl()),
1019+
getURI(context.requestUrl),
10161020
provider);
10171021
context.responseStatus = responseStatus;
10181022
if (context.statusHandler != null) {
@@ -1033,6 +1037,16 @@ protected void onInitialLineParsed(HttpHeader httpHeader,
10331037

10341038
}
10351039

1040+
@Override
1041+
protected void onHttpError(final HttpHeader httpHeader,
1042+
final FilterChainContext ctx,
1043+
final Throwable t) throws IOException {
1044+
httpHeader.setSkipRemainder(true);
1045+
final HttpTransactionContext context =
1046+
provider.getHttpTransactionContext(ctx.getConnection());
1047+
context.abort(t);
1048+
}
1049+
10361050
@SuppressWarnings({"unchecked"})
10371051
@Override
10381052
protected void onHttpHeadersParsed(HttpHeader httpHeader,
@@ -1042,6 +1056,11 @@ protected void onHttpHeadersParsed(HttpHeader httpHeader,
10421056
if (LOGGER.isDebugEnabled()) {
10431057
LOGGER.debug("RESPONSE: " + httpHeader.toString());
10441058
}
1059+
if (httpHeader.containsHeader(Header.Connection)) {
1060+
if ("close".equals(httpHeader.getHeader(Header.Connection))) {
1061+
ConnectionManager.markConnectionAsDoNotCache(ctx.getConnection());
1062+
}
1063+
}
10451064
if (httpHeader.isSkipRemainder()) {
10461065
return;
10471066
}
@@ -1157,20 +1176,21 @@ private static boolean isRedirectAllowed(final HttpTransactionContext ctx) {
11571176
private static HttpTransactionContext cleanup(final FilterChainContext ctx,
11581177
final GrizzlyAsyncHttpProvider provider) {
11591178

1179+
final Connection c = ctx.getConnection();
11601180
final HttpTransactionContext context =
1161-
provider.getHttpTransactionContext(ctx.getConnection());
1162-
1163-
if (!context.provider.connectionManager.canReturnConnection(ctx.getConnection())) {
1181+
provider.getHttpTransactionContext(c);
1182+
context.provider.setHttpTransactionContext(c, null);
1183+
if (!context.provider.connectionManager.canReturnConnection(c)) {
11641184
context.abort(new IOException("Maximum pooled connections exceeded"));
11651185
} else {
1166-
if (!context.provider.connectionManager.returnConnection(context.request.getUrl(), ctx.getConnection())) {
1186+
if (!context.provider.connectionManager.returnConnection(context.requestUrl, c)) {
11671187
try {
11681188
ctx.getConnection().close().markForRecycle(true);
11691189
} catch (IOException ignored) {
11701190
}
11711191
}
11721192
}
1173-
context.provider.setHttpTransactionContext(ctx.getConnection(), null);
1193+
11741194
return context;
11751195

11761196
}
@@ -1237,7 +1257,7 @@ public boolean handleStatus(final HttpResponsePacket responsePacket,
12371257
final Request req = httpTransactionContext.request;
12381258
realm = new Realm.RealmBuilder().clone(realm)
12391259
.setScheme(realm.getAuthScheme())
1240-
.setUri(URI.create(req.getUrl()).getPath())
1260+
.setUri(URI.create(httpTransactionContext.requestUrl).getPath())
12411261
.setMethodName(req.getMethod())
12421262
.setUsePreemptiveAuth(true)
12431263
.parseWWWAuthenticateHeader(auth)
@@ -1314,9 +1334,9 @@ public boolean handleStatus(final HttpResponsePacket responsePacket,
13141334

13151335
URI orig;
13161336
if (httpTransactionContext.lastRedirectURI == null) {
1317-
orig = AsyncHttpProviderUtils.createUri(httpTransactionContext.request.getUrl());
1337+
orig = AsyncHttpProviderUtils.createUri(httpTransactionContext.requestUrl);
13181338
} else {
1319-
orig = AsyncHttpProviderUtils.getRedirectUri(AsyncHttpProviderUtils.createUri(httpTransactionContext.request.getUrl()),
1339+
orig = AsyncHttpProviderUtils.getRedirectUri(AsyncHttpProviderUtils.createUri(httpTransactionContext.requestUrl),
13201340
httpTransactionContext.lastRedirectURI);
13211341
}
13221342
httpTransactionContext.lastRedirectURI = redirectURL;
@@ -1354,6 +1374,7 @@ public boolean handleStatus(final HttpResponsePacket responsePacket,
13541374
httpTransactionContext.future = null;
13551375
newContext.invocationStatus = InvocationStatus.CONTINUE;
13561376
newContext.request = requestToSend;
1377+
newContext.requestUrl = requestToSend.getUrl();
13571378
httpTransactionContext.provider.setHttpTransactionContext(c, newContext);
13581379
httpTransactionContext.provider.execute(c,
13591380
requestToSend,
@@ -1637,7 +1658,7 @@ public void doHandle(final FilterChainContext ctx,
16371658
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
16381659
}
16391660

1640-
} // END StringBodyHandler
1661+
} // END NoBodyHandler
16411662

16421663

16431664
private final class ParamsBodyHandler implements BodyHandler {
@@ -1666,18 +1687,21 @@ public void doHandle(final FilterChainContext ctx,
16661687
charset = Charsets.DEFAULT_CHARACTER_ENCODING;
16671688
}
16681689
final FluentStringsMap params = request.getParams();
1669-
for (Map.Entry<String, List<String>> entry : params.entrySet()) {
1670-
String name = entry.getKey();
1671-
List<String> values = entry.getValue();
1672-
if (values != null && !values.isEmpty()) {
1673-
if (sb == null) {
1674-
sb = new StringBuilder(128);
1675-
}
1676-
for (String value : values) {
1677-
if (sb.length() > 0) {
1678-
sb.append('&');
1690+
if (!params.isEmpty()) {
1691+
for (Map.Entry<String, List<String>> entry : params.entrySet()) {
1692+
String name = entry.getKey();
1693+
List<String> values = entry.getValue();
1694+
if (values != null && !values.isEmpty()) {
1695+
if (sb == null) {
1696+
sb = new StringBuilder(128);
1697+
}
1698+
for (String value : values) {
1699+
if (sb.length() > 0) {
1700+
sb.append('&');
1701+
}
1702+
sb.append(URLEncoder.encode(name, charset))
1703+
.append('=').append(URLEncoder.encode(value, charset));
16791704
}
1680-
sb.append(URLEncoder.encode(name, charset)).append('=').append(URLEncoder.encode(value, charset));
16811705
}
16821706
}
16831707
}
@@ -1954,9 +1978,9 @@ static boolean isConnectionCacheable(final Connection c) {
19541978
return ((canCache != null) ? canCache : false);
19551979
}
19561980

1957-
private void doAsyncTrackedConnection(final Request request,
1958-
final GrizzlyResponseFuture requestFuture,
1959-
final CompletionHandler<Connection> connectHandler)
1981+
void doAsyncTrackedConnection(final Request request,
1982+
final GrizzlyResponseFuture requestFuture,
1983+
final CompletionHandler<Connection> connectHandler)
19601984
throws IOException, ExecutionException, InterruptedException {
19611985
final String url = request.getUrl();
19621986
Connection c = pool.poll(AsyncHttpProviderUtils.getBaseUrl(url));
@@ -1995,7 +2019,7 @@ Connection obtainTrackedConnection(final Request request,
19952019

19962020
Connection obtainConnection(final Request request,
19972021
final GrizzlyResponseFuture requestFuture)
1998-
throws IOException, ExecutionException, InterruptedException {
2022+
throws IOException, ExecutionException, InterruptedException, TimeoutException {
19992023

20002024
final Connection c = (obtainConnection0(request.getUrl(),
20012025
request,
@@ -2005,10 +2029,10 @@ Connection obtainConnection(final Request request,
20052029

20062030
}
20072031

2008-
private void doAsyncConnect(final String url,
2009-
final Request request,
2010-
final GrizzlyResponseFuture requestFuture,
2011-
final CompletionHandler<Connection> connectHandler)
2032+
void doAsyncConnect(final String url,
2033+
final Request request,
2034+
final GrizzlyResponseFuture requestFuture,
2035+
final CompletionHandler<Connection> connectHandler)
20122036
throws IOException, ExecutionException, InterruptedException {
20132037

20142038
final URI uri = AsyncHttpProviderUtils.createUri(url);
@@ -2026,7 +2050,7 @@ private void doAsyncConnect(final String url,
20262050
private Connection obtainConnection0(final String url,
20272051
final Request request,
20282052
final GrizzlyResponseFuture requestFuture)
2029-
throws IOException, ExecutionException, InterruptedException {
2053+
throws IOException, ExecutionException, InterruptedException, TimeoutException {
20302054

20312055
final URI uri = AsyncHttpProviderUtils.createUri(url);
20322056
ProxyServer proxy = getProxyServer(request);
@@ -2035,8 +2059,18 @@ private Connection obtainConnection0(final String url,
20352059
}
20362060
String host = ((proxy != null) ? proxy.getHost() : uri.getHost());
20372061
int port = ((proxy != null) ? proxy.getPort() : uri.getPort());
2038-
return connectionHandler.connect(new InetSocketAddress(host, getPort(uri, port)),
2039-
createConnectionCompletionHandler(request, requestFuture, null)).get();
2062+
int cTimeout = provider.clientConfig.getConnectionTimeoutInMs();
2063+
if (cTimeout > 0) {
2064+
return connectionHandler.connect(new InetSocketAddress(host, getPort(uri, port)),
2065+
createConnectionCompletionHandler(request,
2066+
requestFuture,
2067+
null)).get(cTimeout, TimeUnit.MILLISECONDS);
2068+
} else {
2069+
return connectionHandler.connect(new InetSocketAddress(host, getPort(uri, port)),
2070+
createConnectionCompletionHandler(request,
2071+
requestFuture,
2072+
null)).get();
2073+
}
20402074

20412075
}
20422076

0 commit comments

Comments
 (0)