Fix BodyDeferringAsyncHandler piped streams doc sample, close AsyncHttpClient#1451

Motivation:

Javadoc sample fails because the PipedOutputStream might not connected
to its PipedInputStream counterpart when bodyparts are received.

Modifications:

* Fix javadoc: PipedInputStream must be connected prior to sending the
request.
* Add piped streams based test

Result:

Sample works

Author: slandelle

client/src/main/java/org/asynchttpclient/handler/BodyDeferringAsyncHandler.java

@@ -62,18 +62,20 @@
  * <br>
  * <pre>
  *     PipedOutputStream pout = new PipedOutputStream();
- *     BodyDeferringAsyncHandler bdah = new BodyDeferringAsyncHandler(pout);
- *     // client executes async
- *     Future&lt;Response&gt; fr = client.prepareGet(&quot;http://foo.com/aresource&quot;).execute(bdah);
- *     // main thread will block here until headers are available
- *     Response response = bdah.getResponse();
- *     if (response.getStatusCode() == 200) {
- *      InputStream pin = new BodyDeferringInputStream(fr,new PipedInputStream(pout));
- *      // consume InputStream
- *      ...
- *     } else {
- *      // handle unexpected response status code
- *      ...
+ *     try (PipedInputStream pin = new PipedInputStream(pout)) {
+ *         BodyDeferringAsyncHandler handler = new BodyDeferringAsyncHandler(pout);
+ *         ListenableFuture<&lt;Response&gt; respFut = client.prepareGet(getTargetUrl()).execute(handler);
+ *         Response resp = handler.getResponse();
+ *         // main thread will block here until headers are available
+ *         if (resp.getStatusCode() == 200) {
+ *             try (InputStream is = new BodyDeferringInputStream(respFut, handler, pin)) {
+ *                 // consume InputStream
+ *                 ...
+ *             }
+ *         } else {
+ *             // handle unexpected response status code
+ *             ...
+ *         }
  *     }
  * </pre>
  */

client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java

@@ -21,6 +21,7 @@
 import java.io.OutputStream;
 import java.io.PipedInputStream;
 import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
@@ -29,10 +30,12 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
+import org.apache.commons.io.IOUtils;
 import org.asynchttpclient.AbstractBasicTest;
 import org.asynchttpclient.AsyncHttpClient;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.BoundRequestBuilder;
+import org.asynchttpclient.ListenableFuture;
 import org.asynchttpclient.Response;
 import org.asynchttpclient.handler.BodyDeferringAsyncHandler.BodyDeferringInputStream;
 import org.eclipse.jetty.server.Request;
@@ -112,7 +115,7 @@ public AsyncHttpClientConfig getAsyncHttpClientConfig() {
     @Test(groups = "standalone")
     public void deferredSimple() throws IOException, ExecutionException, TimeoutException, InterruptedException {
         try (AsyncHttpClient client = asyncHttpClient(getAsyncHttpClientConfig())) {
-            BoundRequestBuilder r = client.prepareGet("http://localhost/" + port1 + "/deferredSimple");
+            BoundRequestBuilder r = client.prepareGet(getTargetUrl());
 
             CountingOutputStream cos = new CountingOutputStream();
             BodyDeferringAsyncHandler bdah = new BodyDeferringAsyncHandler(cos);
@@ -136,8 +139,7 @@ public void deferredSimple() throws IOException, ExecutionException, TimeoutExce
     @Test(groups = "standalone", enabled = false)
     public void deferredSimpleWithFailure() throws IOException, ExecutionException, TimeoutException, InterruptedException {
         try (AsyncHttpClient client = asyncHttpClient(getAsyncHttpClientConfig())) {
-            BoundRequestBuilder r = client.prepareGet("http://localhost:" + port1 + "/deferredSimpleWithFailure").addHeader("X-FAIL-TRANSFER",
-                    Boolean.TRUE.toString());
+            BoundRequestBuilder r = client.prepareGet(getTargetUrl()).addHeader("X-FAIL-TRANSFER", Boolean.TRUE.toString());
 
             CountingOutputStream cos = new CountingOutputStream();
             BodyDeferringAsyncHandler bdah = new BodyDeferringAsyncHandler(cos);
@@ -166,7 +168,7 @@ public void deferredSimpleWithFailure() throws IOException, ExecutionException,
     @Test(groups = "standalone")
     public void deferredInputStreamTrick() throws IOException, ExecutionException, TimeoutException, InterruptedException {
         try (AsyncHttpClient client = asyncHttpClient(getAsyncHttpClientConfig())) {
-            BoundRequestBuilder r = client.prepareGet("http://localhost/" + port1 + "/deferredInputStreamTrick");
+            BoundRequestBuilder r = client.prepareGet(getTargetUrl());
 
             PipedOutputStream pos = new PipedOutputStream();
             PipedInputStream pis = new PipedInputStream(pos);
@@ -241,4 +243,26 @@ public void testConnectionRefused() throws IOException, ExecutionException, Time
             bdah.getResponse();
         }
     }
+
+    @Test(groups = "standalone")
+    public void testPipedStreams() throws Exception {
+        try (AsyncHttpClient client = asyncHttpClient(getAsyncHttpClientConfig())) {
+            PipedOutputStream pout = new PipedOutputStream();
+            try (PipedInputStream pin = new PipedInputStream(pout)) {
+                BodyDeferringAsyncHandler handler = new BodyDeferringAsyncHandler(pout);
+                ListenableFuture<Response> respFut = client.prepareGet(getTargetUrl()).execute(handler);
+
+                Response resp = handler.getResponse();
+
+                if (resp.getStatusCode() == 200) {
+                    try (BodyDeferringInputStream is = new BodyDeferringInputStream(respFut, handler, pin)) {
+                        String body = IOUtils.toString(is, StandardCharsets.UTF_8);
+                        assertTrue(body.contains("ABCDEF"));
+                    }
+                } else {
+                    throw new IOException("HTTP error " + resp.getStatusCode());
+                }
+            }
+        }
+    }
 }