Add Async withHandler() support for any response body handler

Author: rbygrave

client/src/main/java/io/avaje/http/client/DHttpAsync.java

@@ -13,12 +13,17 @@ class DHttpAsync implements HttpAsyncResponse {
   }
 
   @Override
-  public CompletableFuture<HttpResponse<Void>> asDiscarding() {
+  public <E> CompletableFuture<HttpResponse<E>> withHandler(HttpResponse.BodyHandler<E>  handler) {
     return request
-      .performSendAsync(false, HttpResponse.BodyHandlers.discarding())
+      .performSendAsync(false, handler)
       .thenApply(request::afterAsync);
   }
 
+  @Override
+  public CompletableFuture<HttpResponse<Void>> asDiscarding() {
+    return withHandler(HttpResponse.BodyHandlers.discarding());
+  }
+
   @Override
   public CompletableFuture<HttpResponse<String>> asString() {
     return request

client/src/main/java/io/avaje/http/client/DHttpClientRequest.java

@@ -553,7 +553,8 @@ public String responseBody() {
       if (encodedResponseBody != null) {
         return context.maxResponseBody(new String(encodedResponseBody.content(), StandardCharsets.UTF_8));
       } else if (httpResponse != null && loggableResponseBody) {
-        return context.maxResponseBody(httpResponse.body().toString());
+        final Object body = httpResponse.body();
+        return (body == null) ? null : context.maxResponseBody(body.toString());
       }
       return null;
     }

client/src/main/java/io/avaje/http/client/HttpAsyncResponse.java

@@ -5,7 +5,7 @@
 import java.util.concurrent.CompletableFuture;
 
 /**
- * Async responses as CompletableFuture.
+ * Async processing of the request with responses as CompletableFuture.
  */
 public interface HttpAsyncResponse {
 
@@ -29,6 +29,8 @@ public interface HttpAsyncResponse {
    *       });
    *
    * }</pre>
+   *
+   * @return The CompletableFuture of the response
    */
   CompletableFuture<HttpResponse<Void>> asDiscarding();
 
@@ -53,9 +55,53 @@ public interface HttpAsyncResponse {
    *       });
    *
    * }</pre>
+   *
+   * @return The CompletableFuture of the response
    */
   CompletableFuture<HttpResponse<String>> asString();
 
+  /**
+   * Process with any given {@code HttpResponse.BodyHandler}.
+   *
+   * <h3>Example: line subscriber</h3>
+   * <p>
+   * Subscribe line by line to the response.
+   * </p>
+   * <pre>{@code
+   *
+   *    CompletableFuture<HttpResponse<Void>> future = clientContext.request()
+   *       .path("hello/lineStream")
+   *       .GET().async()
+   *       .withHandler(HttpResponse.BodyHandlers.fromLineSubscriber(new Flow.Subscriber<>() {
+   *
+   *         @Override
+   *         public void onSubscribe(Flow.Subscription subscription) {
+   *           subscription.request(Long.MAX_VALUE);
+   *         }
+   *         @Override
+   *         public void onNext(String item) {
+   *           ...
+   *         }
+   *         @Override
+   *         public void onError(Throwable throwable) {
+   *           ...
+   *         }
+   *         @Override
+   *         public void onComplete() {
+   *           ...
+   *         }
+   *       }))
+   *       .whenComplete((hres, throwable) -> {
+   *         int statusCode = hres.statusCode();
+   *         ...
+   *       });
+   * }</pre>
+   *
+   * @param bodyHandlers The body handler to use to process the response
+   * @return The CompletableFuture of the response
+   */
+  <E> CompletableFuture<HttpResponse<E>> withHandler(HttpResponse.BodyHandler<E> bodyHandlers);
+
   /**
    * Process expecting a bean response body (typically from json content).
    * <p>
@@ -84,6 +130,9 @@ public interface HttpAsyncResponse {
    *         }
    *       });
    * }</pre>
+   *
+   * @param type The bean type to convert the content to
+   * @return The CompletableFuture of the response
    */
   <E> CompletableFuture<E> bean(Class<E> type);
 
@@ -112,6 +161,9 @@ public interface HttpAsyncResponse {
    *         }
    *       });
    * }</pre>
+   *
+   * @param type The bean type to convert the content to
+   * @return The CompletableFuture of the response
    */
   <E> CompletableFuture<List<E>> list(Class<E> type);
 

client/src/test/java/io/avaje/http/client/HelloControllerTest.java

@@ -5,11 +5,13 @@
 import org.junit.jupiter.api.Test;
 
 import java.net.http.HttpResponse;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
@@ -38,6 +40,55 @@ void get_stream() {
     assertThat(first.name).isEqualTo("one");
   }
 
+  @Test
+  void async_stream() throws ExecutionException, InterruptedException {
+
+    AtomicReference<HttpResponse<Void>> hresRef = new AtomicReference<>();
+    AtomicReference<Throwable> errRef = new AtomicReference<>();
+    AtomicReference<Boolean> completeRef = new AtomicReference<>();
+    AtomicReference<Boolean> onSubscribeRef = new AtomicReference<>();
+
+    final List<String> lines = new ArrayList<>();
+
+    final CompletableFuture<HttpResponse<Void>> future = clientContext.request()
+      .path("hello/stream")
+      .GET()
+      .async().withHandler(HttpResponse.BodyHandlers.fromLineSubscriber(new Flow.Subscriber<>() {
+        @Override
+        public void onSubscribe(Flow.Subscription subscription) {
+          subscription.request(Long.MAX_VALUE);
+          onSubscribeRef.set(true);
+        }
+        @Override
+        public void onNext(String item) {
+          lines.add(item);
+        }
+        @Override
+        public void onError(Throwable throwable) {
+          errRef.set(throwable);
+        }
+        @Override
+        public void onComplete() {
+          completeRef.set(true);
+        }
+      })).whenComplete((hres, throwable) -> {
+        hresRef.set(hres);
+        assertThat(hres.statusCode()).isEqualTo(200);
+        assertThat(throwable).isNull();
+      });
+
+    // just wait
+    assertThat(future.get()).isSameAs(hresRef.get());
+
+    assertThat(onSubscribeRef.get()).isTrue();
+    assertThat(completeRef.get()).isTrue();
+    assertThat(errRef.get()).isNull();
+    assertThat(lines).hasSize(4);
+
+    final String first = lines.get(0);
+    assertThat(first).isEqualTo("{\"id\":1, \"name\":\"one\"}");
+  }
+
   @Test
   void get_helloMessage() {