Skip to content

Commit 55bb178

Browse files
oleksiysOleksiy Stashok
oleksiys
authored and
Oleksiy Stashok
committed
+ refactoring the GrizzlyAsyncHttpProvider to support partial request body transferring.
+ @todo pass context to BodyGenerator so it will be able to continue payload transferring.
1 parent f830861 commit 55bb178

File tree

1 file changed

+58
-26
lines changed

1 file changed

+58
-26
lines changed

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

Lines changed: 58 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,6 @@
9797
import java.net.InetSocketAddress;
9898
import java.net.URI;
9999
import java.net.URLEncoder;
100-
import java.nio.ByteBuffer;
101100
import java.security.NoSuchAlgorithmException;
102101
import java.util.Collection;
103102
import java.util.HashMap;
@@ -460,11 +459,13 @@ static int getPort(final URI uri, final int p) {
460459

461460

462461
@SuppressWarnings({"unchecked"})
463-
void sendRequest(final FilterChainContext ctx,
462+
boolean sendRequest(final FilterChainContext ctx,
464463
final Request request,
465464
final HttpRequestPacket requestPacket)
466465
throws IOException {
467466

467+
boolean isWriteComplete = true;
468+
468469
if (requestHasEntityBody(request)) {
469470
final HttpTransactionContext context = getHttpTransactionContext(ctx.getConnection());
470471
BodyHandler handler = bodyHandlerFactory.getBodyHandler(request);
@@ -473,13 +474,15 @@ void sendRequest(final FilterChainContext ctx,
473474
handler = new ExpectHandler(handler);
474475
}
475476
context.bodyHandler = handler;
476-
handler.doHandle(ctx, request, requestPacket);
477+
isWriteComplete = handler.doHandle(ctx, request, requestPacket);
477478
} else {
478479
ctx.write(requestPacket, ctx.getTransportContext().getCompletionHandler());
479480
}
480481
if (LOGGER.isDebugEnabled()) {
481482
LOGGER.debug("REQUEST: " + requestPacket.toString());
482483
}
484+
485+
return isWriteComplete;
483486
}
484487

485488

@@ -686,7 +689,9 @@ public NextAction handleWrite(final FilterChainContext ctx)
686689
Object message = ctx.getMessage();
687690
if (message instanceof Request) {
688691
ctx.setMessage(null);
689-
sendAsGrizzlyRequest((Request) message, ctx);
692+
if (!sendAsGrizzlyRequest((Request) message, ctx)) {
693+
return ctx.getSuspendAction();
694+
}
690695
}
691696

692697
return ctx.getStopAction();
@@ -713,7 +718,7 @@ public NextAction handleEvent(final FilterChainContext ctx,
713718

714719

715720

716-
private void sendAsGrizzlyRequest(final Request request,
721+
private boolean sendAsGrizzlyRequest(final Request request,
717722
final FilterChainContext ctx)
718723
throws IOException {
719724

@@ -781,7 +786,7 @@ private void sendAsGrizzlyRequest(final Request request,
781786
new FluentCaseInsensitiveStringsMap(request.getHeaders());
782787
TransferCompletionHandler.class.cast(h).transferAdapter(new GrizzlyTransferAdapter(map));
783788
}
784-
sendRequest(ctx, request, requestPacket);
789+
return sendRequest(ctx, request, requestPacket);
785790

786791
}
787792

@@ -1513,7 +1518,7 @@ private static interface BodyHandler {
15131518

15141519
boolean handlesBodyType(final Request request);
15151520

1516-
void doHandle(final FilterChainContext ctx,
1521+
boolean doHandle(final FilterChainContext ctx,
15171522
final Request request,
15181523
final HttpRequestPacket requestPacket) throws IOException;
15191524

@@ -1569,10 +1574,11 @@ public boolean handlesBodyType(Request request) {
15691574
}
15701575

15711576
@SuppressWarnings({"unchecked"})
1572-
public void doHandle(FilterChainContext ctx, Request request, HttpRequestPacket requestPacket) throws IOException {
1577+
public boolean doHandle(FilterChainContext ctx, Request request, HttpRequestPacket requestPacket) throws IOException {
15731578
this.request = request;
15741579
this.requestPacket = requestPacket;
15751580
ctx.write(requestPacket, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
1581+
return true;
15761582
}
15771583

15781584
public void finish(final FilterChainContext ctx) throws IOException {
@@ -1592,7 +1598,7 @@ public boolean handlesBodyType(final Request request) {
15921598
}
15931599

15941600
@SuppressWarnings({"unchecked"})
1595-
public void doHandle(final FilterChainContext ctx,
1601+
public boolean doHandle(final FilterChainContext ctx,
15961602
final Request request,
15971603
final HttpRequestPacket requestPacket)
15981604
throws IOException {
@@ -1612,6 +1618,7 @@ public void doHandle(final FilterChainContext ctx,
16121618
final HttpContent content = requestPacket.httpContentBuilder().content(gBuffer).build();
16131619
content.setLast(true);
16141620
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
1621+
return true;
16151622
}
16161623
}
16171624

@@ -1627,7 +1634,7 @@ public boolean handlesBodyType(final Request request) {
16271634
}
16281635

16291636
@SuppressWarnings({"unchecked"})
1630-
public void doHandle(final FilterChainContext ctx,
1637+
public boolean doHandle(final FilterChainContext ctx,
16311638
final Request request,
16321639
final HttpRequestPacket requestPacket)
16331640
throws IOException {
@@ -1647,6 +1654,7 @@ public void doHandle(final FilterChainContext ctx,
16471654
final HttpContent content = requestPacket.httpContentBuilder().content(gBuffer).build();
16481655
content.setLast(true);
16491656
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
1657+
return true;
16501658
}
16511659

16521660
} // END StringBodyHandler
@@ -1663,14 +1671,15 @@ public boolean handlesBodyType(final Request request) {
16631671
}
16641672

16651673
@SuppressWarnings({"unchecked"})
1666-
public void doHandle(final FilterChainContext ctx,
1674+
public boolean doHandle(final FilterChainContext ctx,
16671675
final Request request,
16681676
final HttpRequestPacket requestPacket)
16691677
throws IOException {
16701678

16711679
final HttpContent content = requestPacket.httpContentBuilder().content(Buffers.EMPTY_BUFFER).build();
16721680
content.setLast(true);
16731681
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
1682+
return true;
16741683
}
16751684

16761685
} // END NoBodyHandler
@@ -1688,7 +1697,7 @@ public boolean handlesBodyType(final Request request) {
16881697
}
16891698

16901699
@SuppressWarnings({"unchecked"})
1691-
public void doHandle(final FilterChainContext ctx,
1700+
public boolean doHandle(final FilterChainContext ctx,
16921701
final Request request,
16931702
final HttpRequestPacket requestPacket)
16941703
throws IOException {
@@ -1733,6 +1742,7 @@ public void doHandle(final FilterChainContext ctx,
17331742
content.setLast(true);
17341743
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
17351744
}
1745+
return true;
17361746
}
17371747

17381748
} // END ParamsBodyHandler
@@ -1748,7 +1758,7 @@ public boolean handlesBodyType(final Request request) {
17481758
}
17491759

17501760
@SuppressWarnings({"unchecked"})
1751-
public void doHandle(final FilterChainContext ctx,
1761+
public boolean doHandle(final FilterChainContext ctx,
17521762
final Request request,
17531763
final HttpRequestPacket requestPacket)
17541764
throws IOException {
@@ -1766,6 +1776,7 @@ public void doHandle(final FilterChainContext ctx,
17661776
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
17671777
}
17681778

1779+
return true;
17691780
}
17701781

17711782
} // END EntityWriterBodyHandler
@@ -1781,7 +1792,7 @@ public boolean handlesBodyType(final Request request) {
17811792
}
17821793

17831794
@SuppressWarnings({"unchecked"})
1784-
public void doHandle(final FilterChainContext ctx,
1795+
public boolean doHandle(final FilterChainContext ctx,
17851796
final Request request,
17861797
final HttpRequestPacket requestPacket)
17871798
throws IOException {
@@ -1815,6 +1826,8 @@ public void doHandle(final FilterChainContext ctx,
18151826
content.setLast(true);
18161827
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
18171828
}
1829+
1830+
return true;
18181831
}
18191832

18201833
} // END StreamDataBodyHandler
@@ -1831,7 +1844,7 @@ public boolean handlesBodyType(final Request request) {
18311844
}
18321845

18331846
@SuppressWarnings({"unchecked"})
1834-
public void doHandle(final FilterChainContext ctx,
1847+
public boolean doHandle(final FilterChainContext ctx,
18351848
final Request request,
18361849
final HttpRequestPacket requestPacket)
18371850
throws IOException {
@@ -1854,6 +1867,7 @@ public void doHandle(final FilterChainContext ctx,
18541867
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
18551868
}
18561869

1870+
return true;
18571871
}
18581872

18591873
} // END PartsBodyHandler
@@ -1869,7 +1883,7 @@ public boolean handlesBodyType(final Request request) {
18691883
}
18701884

18711885
@SuppressWarnings({"unchecked"})
1872-
public void doHandle(final FilterChainContext ctx,
1886+
public boolean doHandle(final FilterChainContext ctx,
18731887
final Request request,
18741888
final HttpRequestPacket requestPacket)
18751889
throws IOException {
@@ -1897,6 +1911,8 @@ public void doHandle(final FilterChainContext ctx,
18971911
last(last).build();
18981912
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
18991913
}
1914+
1915+
return true;
19001916
}
19011917

19021918
} // END FileBodyHandler
@@ -1912,7 +1928,7 @@ public boolean handlesBodyType(final Request request) {
19121928
}
19131929

19141930
@SuppressWarnings({"unchecked"})
1915-
public void doHandle(final FilterChainContext ctx,
1931+
public boolean doHandle(final FilterChainContext ctx,
19161932
final Request request,
19171933
final HttpRequestPacket requestPacket)
19181934
throws IOException {
@@ -1926,21 +1942,37 @@ public void doHandle(final FilterChainContext ctx,
19261942
requestPacket.setChunked(true);
19271943
}
19281944

1945+
final MemoryManager mm = ctx.getMemoryManager();
19291946
boolean last = false;
1930-
for (ByteBuffer buffer = ByteBuffer.allocate(MAX_CHUNK_SIZE); !last; ) {
1931-
buffer.clear();
1932-
if (bodyLocal.read(buffer) < 0) {
1933-
last = true;
1934-
buffer = ByteBuffer.allocate(0);
1947+
1948+
while (!last) {
1949+
Buffer buffer = mm.allocate(MAX_CHUNK_SIZE);
1950+
buffer.allowBufferDispose(true);
1951+
1952+
final long readBytes = bodyLocal.read(buffer.toByteBuffer());
1953+
if (readBytes > 0) {
1954+
buffer.position((int) readBytes);
1955+
buffer.trim();
1956+
} else {
1957+
buffer.dispose();
1958+
1959+
if (readBytes < 0) {
1960+
last = true;
1961+
buffer = Buffers.EMPTY_BUFFER;
1962+
} else {
1963+
// @TODO pass the context to bodyLocal to be able to
1964+
// continue body transferring once more data is available
1965+
return false;
1966+
}
19351967
}
1936-
final MemoryManager mm = ctx.getMemoryManager();
1937-
buffer.flip();
1938-
Buffer b = Buffers.wrap(mm, buffer);
1968+
19391969
final HttpContent content =
1940-
requestPacket.httpContentBuilder().content(b).
1970+
requestPacket.httpContentBuilder().content(buffer).
19411971
last(last).build();
19421972
ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
19431973
}
1974+
1975+
return true;
19441976
}
19451977

19461978
} // END BodyGeneratorBodyHandler

0 commit comments

Comments
 (0)