Add built-in support for InputStream body with defined Content-Length (AsyncHttpClient#1282)

Current state:
Hi, now if you pass InputStream as a body source - you always get request with chunked encoding.

Motivation:
Provide built-in ability to send requests with static Content-Length if we have fully in-memory InputStream (like ByteArrayInputStream) + length.

Changes:
Adopt InputStreamBodyGenerator and supporting code to work with configurable contentLength.

Author: Spikhalskiy

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

@@ -103,7 +103,8 @@ private NettyBody body(Request request, boolean connect) {
                 nettyBody = new NettyFileBody(fileBodyGenerator.getFile(), fileBodyGenerator.getRegionSeek(), fileBodyGenerator.getRegionLength(), config);
 
             } else if (request.getBodyGenerator() instanceof InputStreamBodyGenerator) {
-                nettyBody = new NettyInputStreamBody(InputStreamBodyGenerator.class.cast(request.getBodyGenerator()).getInputStream());
+                InputStreamBodyGenerator inStreamGenerator = InputStreamBodyGenerator.class.cast(request.getBodyGenerator());
+                nettyBody = new NettyInputStreamBody(inStreamGenerator.getInputStream(), inStreamGenerator.getContentLength());
 
             } else if (request.getBodyGenerator() instanceof ReactiveStreamsBodyGenerator) {
                 ReactiveStreamsBodyGenerator reactiveStreamsBodyGenerator = (ReactiveStreamsBodyGenerator)request.getBodyGenerator();

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

@@ -33,9 +33,15 @@ public class NettyInputStreamBody implements NettyBody {
     private static final Logger LOGGER = LoggerFactory.getLogger(NettyInputStreamBody.class);
 
     private final InputStream inputStream;
+    private final long contentLength;
 
     public NettyInputStreamBody(InputStream inputStream) {
+        this(inputStream, -1L);
+    }
+
+    public NettyInputStreamBody(InputStream inputStream, long contentLength) {
         this.inputStream = inputStream;
+        this.contentLength = contentLength;
     }
 
     public InputStream getInputStream() {
@@ -44,7 +50,7 @@ public InputStream getInputStream() {
 
     @Override
     public long getContentLength() {
-        return -1L;
+        return contentLength;
     }
 
     @Override

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

@@ -32,34 +32,46 @@ public final class InputStreamBodyGenerator implements BodyGenerator {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(InputStreamBody.class);
     private final InputStream inputStream;
+    private final long contentLength;
 
     public InputStreamBodyGenerator(InputStream inputStream) {
+        this(inputStream, -1L);
+    }
+
+    public InputStreamBodyGenerator(InputStream inputStream, long contentLength) {
         this.inputStream = inputStream;
+        this.contentLength = contentLength;
     }
 
     public InputStream getInputStream() {
         return inputStream;
     }
 
+    public long getContentLength() {
+        return contentLength;
+    }
+
     /**
      * {@inheritDoc}
      */
     @Override
     public Body createBody() {
-        return new InputStreamBody(inputStream);
+        return new InputStreamBody(inputStream, contentLength);
     }
 
     private class InputStreamBody implements Body {
 
         private final InputStream inputStream;
+        private final long contentLength;
         private byte[] chunk;
 
-        private InputStreamBody(InputStream inputStream) {
+        private InputStreamBody(InputStream inputStream, long contentLength) {
             this.inputStream = inputStream;
+            this.contentLength = contentLength;
         }
 
         public long getContentLength() {
-            return -1L;
+            return contentLength;
         }
 
         public BodyState transferTo(ByteBuf target) throws IOException {

client/src/test/java/org/asynchttpclient/BasicHttpTest.java

@@ -25,6 +25,8 @@
 import io.netty.handler.codec.http.HttpHeaders;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
 import java.net.ConnectException;
 import java.net.UnknownHostException;
 import java.nio.charset.StandardCharsets;
@@ -42,14 +44,20 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import javax.net.ssl.SSLException;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
 
 import org.asynchttpclient.cookie.Cookie;
 import org.asynchttpclient.handler.MaxRedirectException;
+import org.asynchttpclient.request.body.generator.InputStreamBodyGenerator;
 import org.asynchttpclient.request.body.multipart.StringPart;
 import org.asynchttpclient.test.EventCollectingHandler;
 import org.asynchttpclient.test.TestUtils.AsyncCompletionHandlerAdapter;
 import org.asynchttpclient.testserver.HttpServer;
+import org.asynchttpclient.testserver.HttpServer.EchoHandler;
 import org.asynchttpclient.testserver.HttpTest;
+import org.eclipse.jetty.server.handler.AbstractHandler;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -911,4 +919,76 @@ public void requestingPlainHttpEndpointOverHttpsThrowsSslException() throws Thro
             });
         });
     }
+
+    @Test
+    public void postUnboundedInputStreamAsBodyStream() throws Throwable {
+        withClient().run(client -> {
+            withServer(server).run(server -> {
+                HttpHeaders h = new DefaultHttpHeaders();
+                h.add(CONTENT_TYPE, APPLICATION_JSON);
+                server.enqueue(new AbstractHandler() {
+                    EchoHandler chain = new EchoHandler();
+                    @Override
+                    public void handle(String target, org.eclipse.jetty.server.Request request,
+                                       HttpServletRequest httpServletRequest,
+                                       HttpServletResponse httpServletResponse) throws IOException, ServletException {
+                        assertEquals(request.getHeader(TRANSFER_ENCODING), CHUNKED);
+                        assertNull(request.getHeader(CONTENT_LENGTH));
+                        chain.handle(target, request, httpServletRequest, httpServletResponse);
+                    }
+                });
+                server.enqueueEcho();
+
+                client.preparePost(getTargetUrl())//
+                      .setHeaders(h)//
+                      .setBody(new ByteArrayInputStream("{}".getBytes(StandardCharsets.ISO_8859_1)))//
+                      .execute(new AsyncCompletionHandlerAdapter() {
+                          @Override
+                          public Response onCompleted(Response response) throws Exception {
+                              assertEquals(response.getStatusCode(), 200);
+                              assertEquals(response.getResponseBody(), "{}");
+                              return response;
+                          }
+                      }).get(TIMEOUT, SECONDS);
+            });
+        });
+    }
+
+    @Test
+    public void postInputStreamWithContentLengthAsBodyGenerator() throws Throwable {
+        withClient().run(client -> {
+            withServer(server).run(server -> {
+                HttpHeaders h = new DefaultHttpHeaders();
+                h.add(CONTENT_TYPE, APPLICATION_JSON);
+                server.enqueue(new AbstractHandler() {
+                    EchoHandler chain = new EchoHandler();
+                    @Override
+                    public void handle(String target, org.eclipse.jetty.server.Request request,
+                                       HttpServletRequest httpServletRequest,
+                                       HttpServletResponse httpServletResponse) throws IOException, ServletException {
+                        assertNull(request.getHeader(TRANSFER_ENCODING));
+                        assertEquals(request.getHeader(CONTENT_LENGTH),//
+                                     Integer.toString("{}".getBytes(StandardCharsets.ISO_8859_1).length));
+                        chain.handle(target, request, httpServletRequest, httpServletResponse);
+                    }
+                });
+
+                byte[] bodyBytes = "{}".getBytes(StandardCharsets.ISO_8859_1);
+                InputStream bodyStream = new ByteArrayInputStream(bodyBytes);
+
+                client.preparePost(getTargetUrl())//
+                      .setHeaders(h)//
+                      .setBody(new InputStreamBodyGenerator(bodyStream, bodyBytes.length))//
+                      .execute(new AsyncCompletionHandlerAdapter() {
+
+                          @Override
+                          public Response onCompleted(Response response) throws Exception {
+                              assertEquals(response.getStatusCode(), 200);
+                              assertEquals(response.getResponseBody(), "{}");
+                              return response;
+                          }
+                      }).get(TIMEOUT, SECONDS);
+            });
+        });
+    }
 }