Merge pull request AsyncHttpClient#1139 from schmitch/stream-with-content-length

Added a way to define a body for ReactiveStreams, close AsyncHttpClient#1136

Author: slandelle

client/src/main/java/org/asynchttpclient/RequestBuilderBase.java

@@ -308,7 +308,11 @@ public T setBody(InputStream stream) {
     }
 
     public T setBody(Publisher<ByteBuffer> publisher) {
-        return setBody(new ReactiveStreamsBodyGenerator(publisher));
+        return setBody(publisher, -1L);
+    }
+
+    public T setBody(Publisher<ByteBuffer> publisher, long contentLength) {
+        return setBody(new ReactiveStreamsBodyGenerator(publisher, contentLength));
     }
 
     public T setBody(BodyGenerator bodyGenerator) {

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java

@@ -103,7 +103,8 @@ private NettyBody body(Request request, boolean connect) {
                 nettyBody = new NettyInputStreamBody(InputStreamBodyGenerator.class.cast(request.getBodyGenerator()).getInputStream());
 
             } else if (request.getBodyGenerator() instanceof ReactiveStreamsBodyGenerator) {
-                nettyBody = new NettyReactiveStreamsBody(ReactiveStreamsBodyGenerator.class.cast(request.getBodyGenerator()).getPublisher());
+                ReactiveStreamsBodyGenerator reactiveStreamsBodyGenerator = (ReactiveStreamsBodyGenerator)request.getBodyGenerator();
+                nettyBody = new NettyReactiveStreamsBody(reactiveStreamsBodyGenerator.getPublisher(), reactiveStreamsBodyGenerator.getContentLength());
 
             } else if (request.getBodyGenerator() != null) {
                 nettyBody = new NettyBodyBody(request.getBodyGenerator().createBody(), config);

client/src/main/java/org/asynchttpclient/netty/request/body/NettyReactiveStreamsBody.java

@@ -42,13 +42,16 @@ public class NettyReactiveStreamsBody implements NettyBody {
 
     private final Publisher<ByteBuffer> publisher;
 
-    public NettyReactiveStreamsBody(Publisher<ByteBuffer> publisher) {
+    private final long contentLength;
+
+    public NettyReactiveStreamsBody(Publisher<ByteBuffer> publisher, long contentLength) {
         this.publisher = publisher;
+        this.contentLength = contentLength;
     }
 
     @Override
     public long getContentLength() {
-        return -1L;
+        return contentLength;
     }
 
     @Override

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

@@ -31,10 +31,20 @@ public class ReactiveStreamsBodyGenerator implements FeedableBodyGenerator {
     private final Publisher<ByteBuffer> publisher;
     private final FeedableBodyGenerator feedableBodyGenerator;
     private volatile FeedListener feedListener;
-
-    public ReactiveStreamsBodyGenerator(Publisher<ByteBuffer> publisher) {
+    private final long contentLength;
+
+    /**
+     *  Creates a Streamable Body which takes a Content-Length.
+     *  If the contentLength parameter is -1L a Http Header of Transfer-Encoding: chunked will be set.
+     *  Otherwise it will set the Content-Length header to the value provided
+     *
+     * @param publisher Body as a Publisher
+     * @param contentLength Content-Length of the Body
+     */
+    public ReactiveStreamsBodyGenerator(Publisher<ByteBuffer> publisher, long contentLength) {
         this.publisher = publisher;
         this.feedableBodyGenerator = new UnboundedQueueFeedableBodyGenerator();
+        this.contentLength = contentLength;
     }
 
     public Publisher<ByteBuffer> getPublisher() {
@@ -52,9 +62,13 @@ public void setListener(FeedListener listener) {
         feedableBodyGenerator.setListener(listener);
     }
 
+    public long getContentLength() {
+        return contentLength;
+    }
+
     @Override
     public Body createBody() {
-        return new StreamedBody(publisher, feedableBodyGenerator);
+        return new StreamedBody(publisher, feedableBodyGenerator, contentLength);
     }
 
     private class StreamedBody implements Body {
@@ -63,9 +77,12 @@ private class StreamedBody implements Body {
         private final SimpleSubscriber subscriber;
         private final Body body;
 
-        public StreamedBody(Publisher<ByteBuffer> publisher, FeedableBodyGenerator bodyGenerator) {
+        private final long contentLength;
+
+        public StreamedBody(Publisher<ByteBuffer> publisher, FeedableBodyGenerator bodyGenerator, long contentLength) {
             this.body = bodyGenerator.createBody();
             this.subscriber = new SimpleSubscriber(bodyGenerator);
+            this.contentLength = contentLength;
         }
 
         @Override
@@ -75,7 +92,7 @@ public void close() throws IOException {
 
         @Override
         public long getContentLength() {
-            return body.getContentLength();
+            return contentLength;
         }
 
         @Override