Skip to content

Commit da99f3e

Browse files
slandelleslandelle
committed
Drop FeedableBodyGenerator.writeChunkBoundaries that was for Netty 3
1 parent f07f0a9 commit da99f3e

File tree

7 files changed

+58
-194
lines changed

7 files changed

+58
-194
lines changed

‎client/src/main/java/org/asynchttpclient/request/body/generator/FeedableBodyGenerator.java

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,6 @@
2222
public interface FeedableBodyGenerator extends BodyGenerator {
2323
void feed(ByteBuffer buffer, boolean isLast);
2424

25-
void writeChunkBoundaries();
26-
2725
void setListener(FeedListener listener);
2826

2927
interface FeedListener {

‎client/src/main/java/org/asynchttpclient/request/body/generator/ReactiveStreamsBodyGenerator.java

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -45,11 +45,6 @@ public void feed(ByteBuffer buffer, boolean isLast) {
4545
feedableBodyGenerator.feed(buffer, isLast);
4646
}
4747

48-
@Override
49-
public void writeChunkBoundaries() {
50-
feedableBodyGenerator.writeChunkBoundaries();
51-
}
52-
5348
@Override
5449
public void setListener(FeedListener listener) {
5550
feedListener.set(listener);

‎client/src/main/java/org/asynchttpclient/request/body/generator/SimpleFeedableBodyGenerator.java

Lines changed: 1 addition & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,6 @@
1313
*/
1414
package org.asynchttpclient.request.body.generator;
1515

16-
import static java.nio.charset.StandardCharsets.US_ASCII;
17-
1816
import java.io.IOException;
1917
import java.nio.ByteBuffer;
2018
import java.util.Queue;
@@ -23,15 +21,9 @@
2321
import org.asynchttpclient.request.body.Body;
2422

2523
public final class SimpleFeedableBodyGenerator implements FeedableBodyGenerator, BodyGenerator {
26-
private final static byte[] END_PADDING = "\r\n".getBytes(US_ASCII);
27-
private final static byte[] ZERO = "0".getBytes(US_ASCII);
28-
private final static ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
2924
private final Queue<BodyPart> queue = new ConcurrentLinkedQueue<>();
3025
private FeedListener listener;
3126

32-
// must be set to true when using Netty 3 where native chunking is broken
33-
private boolean writeChunkBoundaries = false;
34-
3527
@Override
3628
public Body createBody() {
3729
return new PushBody();
@@ -50,11 +42,6 @@ public void setListener(FeedListener listener) {
5042
this.listener = listener;
5143
}
5244

53-
@Override
54-
public void writeChunkBoundaries() {
55-
this.writeChunkBoundaries = true;
56-
}
57-
5845
public final class PushBody implements Body {
5946

6047
private BodyState state = BodyState.CONTINUE;
@@ -95,12 +82,9 @@ private BodyState readNextPart(ByteBuffer buffer) throws IOException {
9582
}
9683

9784
private void readBodyPart(ByteBuffer buffer, BodyPart part) {
98-
part.initBoundaries();
99-
move(buffer, part.size);
10085
move(buffer, part.buffer);
101-
move(buffer, part.endPadding);
10286

103-
if (!part.buffer.hasRemaining() && !part.endPadding.hasRemaining()) {
87+
if (!part.buffer.hasRemaining()) {
10488
if (part.isLast) {
10589
state = BodyState.STOP;
10690
}
@@ -125,47 +109,11 @@ private void move(ByteBuffer destination, ByteBuffer source) {
125109

126110
private final class BodyPart {
127111
private final boolean isLast;
128-
private ByteBuffer size = null;
129112
private final ByteBuffer buffer;
130-
private ByteBuffer endPadding = null;
131113

132114
public BodyPart(final ByteBuffer buffer, final boolean isLast) {
133115
this.buffer = buffer;
134116
this.isLast = isLast;
135117
}
136-
137-
private void initBoundaries() {
138-
if(size == null && endPadding == null) {
139-
if (SimpleFeedableBodyGenerator.this.writeChunkBoundaries) {
140-
if(buffer.hasRemaining()) {
141-
final byte[] sizeAsHex = Integer.toHexString(buffer.remaining()).getBytes(US_ASCII);
142-
size = ByteBuffer.allocate(sizeAsHex.length + END_PADDING.length);
143-
size.put(sizeAsHex);
144-
size.put(END_PADDING);
145-
size.flip();
146-
} else {
147-
size = EMPTY_BUFFER;
148-
}
149-
150-
if(isLast) {
151-
endPadding = ByteBuffer.allocate(END_PADDING.length * 3 + ZERO.length);
152-
if(buffer.hasRemaining()) {
153-
endPadding.put(END_PADDING);
154-
}
155-
156-
//add last empty
157-
endPadding.put(ZERO);
158-
endPadding.put(END_PADDING);
159-
endPadding.put(END_PADDING);
160-
endPadding.flip();
161-
} else {
162-
endPadding = ByteBuffer.wrap(END_PADDING);
163-
}
164-
} else {
165-
size = EMPTY_BUFFER;
166-
endPadding = EMPTY_BUFFER;
167-
}
168-
}
169-
}
170118
}
171119
}

‎client/src/test/java/org/asynchttpclient/reactivestreams/ReactiveStreamsTest.java

Lines changed: 55 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -13,50 +13,89 @@
1313
package org.asynchttpclient.reactivestreams;
1414

1515
import static org.asynchttpclient.Dsl.*;
16-
import static org.asynchttpclient.test.TestUtils.LARGE_IMAGE_BYTES;
16+
import static org.asynchttpclient.test.TestUtils.*;
1717
import static org.testng.Assert.assertEquals;
1818

1919
import java.io.ByteArrayOutputStream;
20+
import java.nio.ByteBuffer;
2021
import java.util.ArrayList;
2122
import java.util.Collections;
2223
import java.util.List;
2324
import java.util.concurrent.CountDownLatch;
25+
import java.util.concurrent.ExecutionException;
2426

2527
import org.asynchttpclient.AbstractBasicTest;
2628
import org.asynchttpclient.AsyncHttpClient;
29+
import org.asynchttpclient.BoundRequestBuilder;
2730
import org.asynchttpclient.HttpResponseBodyPart;
2831
import org.asynchttpclient.HttpResponseHeaders;
2932
import org.asynchttpclient.HttpResponseStatus;
3033
import org.asynchttpclient.ListenableFuture;
34+
import org.asynchttpclient.Response;
3135
import org.asynchttpclient.handler.StreamedAsyncHandler;
3236
import org.reactivestreams.Publisher;
3337
import org.reactivestreams.Subscriber;
3438
import org.reactivestreams.Subscription;
3539
import org.testng.annotations.Test;
3640

41+
import rx.Observable;
42+
import rx.RxReactiveStreams;
43+
3744
public class ReactiveStreamsTest extends AbstractBasicTest {
3845

46+
@Test(groups = { "standalone", "default_provider" })
47+
public void testStreamingPutImage() throws Exception {
48+
try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
49+
Response response = client.preparePut(getTargetUrl()).setBody(LARGE_IMAGE_PUBLISHER).execute().get();
50+
assertEquals(response.getStatusCode(), 200);
51+
assertEquals(response.getResponseBodyAsBytes(), LARGE_IMAGE_BYTES);
52+
}
53+
}
54+
55+
@Test(groups = { "standalone", "default_provider" })
56+
public void testConnectionDoesNotGetClosed() throws Exception {
57+
// test that we can stream the same request multiple times
58+
try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
59+
BoundRequestBuilder requestBuilder = client.preparePut(getTargetUrl()).setBody(LARGE_IMAGE_PUBLISHER);
60+
Response response = requestBuilder.execute().get();
61+
assertEquals(response.getStatusCode(), 200);
62+
assertEquals(response.getResponseBodyAsBytes(), LARGE_IMAGE_BYTES);
63+
64+
response = requestBuilder.execute().get();
65+
assertEquals(response.getStatusCode(), 200);
66+
assertEquals(response.getResponseBodyAsBytes(), LARGE_IMAGE_BYTES);
67+
}
68+
}
69+
70+
@Test(groups = { "standalone", "default_provider" }, expectedExceptions = ExecutionException.class)
71+
public void testFailingStream() throws Exception {
72+
try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
73+
Observable<ByteBuffer> failingObservable = Observable.error(new FailedStream());
74+
Publisher<ByteBuffer> failingPublisher = RxReactiveStreams.toPublisher(failingObservable);
75+
76+
client.preparePut(getTargetUrl()).setBody(failingPublisher).execute().get();
77+
}
78+
}
79+
80+
@SuppressWarnings("serial")
81+
private class FailedStream extends RuntimeException {
82+
}
83+
3984
@Test(groups = { "standalone", "default_provider" })
4085
public void streamedResponseTest() throws Throwable {
4186
try (AsyncHttpClient c = asyncHttpClient()) {
4287

43-
ListenableFuture<SimpleStreamedAsyncHandler> future = c.preparePost(getTargetUrl())
44-
.setBody(LARGE_IMAGE_BYTES)
45-
.execute(new SimpleStreamedAsyncHandler());
88+
ListenableFuture<SimpleStreamedAsyncHandler> future = c.preparePost(getTargetUrl()).setBody(LARGE_IMAGE_BYTES).execute(new SimpleStreamedAsyncHandler());
4689

4790
assertEquals(future.get().getBytes(), LARGE_IMAGE_BYTES);
4891

4992
// Run it again to check that the pipeline is in a good state
50-
future = c.preparePost(getTargetUrl())
51-
.setBody(LARGE_IMAGE_BYTES)
52-
.execute(new SimpleStreamedAsyncHandler());
93+
future = c.preparePost(getTargetUrl()).setBody(LARGE_IMAGE_BYTES).execute(new SimpleStreamedAsyncHandler());
5394

5495
assertEquals(future.get().getBytes(), LARGE_IMAGE_BYTES);
5596

5697
// Make sure a regular request still works
57-
assertEquals(c.preparePost(getTargetUrl())
58-
.setBody("Hello")
59-
.execute().get().getResponseBody(), "Hello");
98+
assertEquals(c.preparePost(getTargetUrl()).setBody("Hello").execute().get().getResponseBody(), "Hello");
6099

61100
}
62101
}
@@ -66,29 +105,21 @@ public void cancelStreamedResponseTest() throws Throwable {
66105
try (AsyncHttpClient c = asyncHttpClient()) {
67106

68107
// Cancel immediately
69-
c.preparePost(getTargetUrl())
70-
.setBody(LARGE_IMAGE_BYTES)
71-
.execute(new CancellingStreamedAsyncProvider(0)).get();
108+
c.preparePost(getTargetUrl()).setBody(LARGE_IMAGE_BYTES).execute(new CancellingStreamedAsyncProvider(0)).get();
72109

73110
// Cancel after 1 element
74-
c.preparePost(getTargetUrl())
75-
.setBody(LARGE_IMAGE_BYTES)
76-
.execute(new CancellingStreamedAsyncProvider(1)).get();
111+
c.preparePost(getTargetUrl()).setBody(LARGE_IMAGE_BYTES).execute(new CancellingStreamedAsyncProvider(1)).get();
77112

78113
// Cancel after 10 elements
79-
c.preparePost(getTargetUrl())
80-
.setBody(LARGE_IMAGE_BYTES)
81-
.execute(new CancellingStreamedAsyncProvider(10)).get();
114+
c.preparePost(getTargetUrl()).setBody(LARGE_IMAGE_BYTES).execute(new CancellingStreamedAsyncProvider(10)).get();
82115

83116
// Make sure a regular request works
84-
assertEquals(c.preparePost(getTargetUrl())
85-
.setBody("Hello")
86-
.execute().get().getResponseBody(), "Hello");
117+
assertEquals(c.preparePost(getTargetUrl()).setBody("Hello").execute().get().getResponseBody(), "Hello");
87118

88119
}
89120
}
90121

91-
static protected class SimpleStreamedAsyncHandler implements StreamedAsyncHandler<SimpleStreamedAsyncHandler>{
122+
static protected class SimpleStreamedAsyncHandler implements StreamedAsyncHandler<SimpleStreamedAsyncHandler> {
92123
private final SimpleSubscriber<HttpResponseBodyPart> subscriber;
93124

94125
public SimpleStreamedAsyncHandler() {
@@ -98,6 +129,7 @@ public SimpleStreamedAsyncHandler() {
98129
public SimpleStreamedAsyncHandler(SimpleSubscriber<HttpResponseBodyPart> subscriber) {
99130
this.subscriber = subscriber;
100131
}
132+
101133
@Override
102134
public State onStream(Publisher<HttpResponseBodyPart> publisher) {
103135
publisher.subscribe(subscriber);

‎client/src/test/java/org/asynchttpclient/request/body/ChunkingTest.java

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -35,11 +35,6 @@
3535
import org.asynchttpclient.request.body.generator.SimpleFeedableBodyGenerator;
3636
import org.testng.annotations.Test;
3737

38-
/**
39-
* Test that the url fetcher is able to communicate via a proxy
40-
*
41-
* @author dominict
42-
*/
4338
public class ChunkingTest extends AbstractBasicTest {
4439
// So we can just test the returned data is the image,
4540
// and doesn't contain the chunked delimeters.

‎client/src/test/java/org/asynchttpclient/request/body/ReactiveStreamsTest.java

Lines changed: 0 additions & 69 deletions
This file was deleted.

0 commit comments

Comments
 (0)