#20 - ENH: Add async().stream(...) for streaming results to beans (typically x-json-stream response)

Author: rbygrave

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

@@ -3,6 +3,7 @@
 import java.net.http.HttpResponse;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
 
 class DHttpAsync implements HttpAsyncResponse {
 
@@ -44,4 +45,11 @@ public <E> CompletableFuture<List<E>> list(Class<E> type) {
       .performSendAsync(true, HttpResponse.BodyHandlers.ofByteArray())
       .thenApply(httpResponse -> request.asyncList(type, httpResponse));
   }
+
+  @Override
+  public <E> CompletableFuture<Stream<E>> stream(Class<E> type) {
+    return request
+      .performSendAsync(false, HttpResponse.BodyHandlers.ofLines())
+      .thenApply(httpResponse -> request.asyncStream(type, httpResponse));
+  }
 }

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

@@ -409,6 +409,17 @@ protected <E> List<E> asyncList(Class<E> type, HttpResponse<byte[]> response) {
     return context.readList(type, encodedResponseBody);
   }
 
+  protected <E> Stream<E> asyncStream(Class<E> type, HttpResponse<Stream<String>> response) {
+    requestTimeNanos = System.nanoTime() - startAsyncNanos;
+    httpResponse = response;
+    context.afterResponse(this);
+    if (response.statusCode() >= 300) {
+      throw new HttpException(response, context);
+    }
+    final BodyReader<E> bodyReader = context.beanReader(type);
+    return response.body().map(bodyReader::readBody);
+  }
+
   private void afterAsyncEncoded(HttpResponse<byte[]> response) {
     requestTimeNanos = System.nanoTime() - startAsyncNanos;
     httpResponse = response;

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

@@ -3,6 +3,7 @@
 import java.net.http.HttpResponse;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
 
 /**
  * Async processing of the request with responses as CompletableFuture.
@@ -105,8 +106,9 @@ public interface HttpAsyncResponse {
   /**
    * Process expecting a bean response body (typically from json content).
    * <p>
-   * If the HTTP statusCode is 300 or above a HttpException is throw which
-   * contains the HttpResponse.
+   * If the HTTP statusCode is 300 or above a HttpException is throw
+   * which contains the HttpResponse. This is the cause in the
+   * CompletionException.
    *
    * <pre>{@code
    *
@@ -125,7 +127,7 @@ public interface HttpAsyncResponse {
    *           ..
    *
    *         } else {
-   *           // use helloDto
+   *           // process helloDto
    *           ...
    *         }
    *       });
@@ -139,8 +141,9 @@ public interface HttpAsyncResponse {
   /**
    * Process expecting a list of beans response body (typically from json content).
    * <p>
-   * If the HTTP statusCode is 300 or above a HttpException is throw which
-   * contains the HttpResponse.
+   * If the HTTP statusCode is 300 or above a HttpException is throw
+   * which contains the HttpResponse. This is the cause in the
+   * CompletionException.
    *
    * <pre>{@code
    *
@@ -156,7 +159,7 @@ public interface HttpAsyncResponse {
    *           ...
    *
    *         } else {
-   *           // use list of helloDto
+   *           // process list of helloDto
    *           ...
    *         }
    *       });
@@ -167,4 +170,35 @@ public interface HttpAsyncResponse {
    */
   <E> CompletableFuture<List<E>> list(Class<E> type);
 
+  /**
+   * Process response as a stream of beans (x-json-stream).
+   * <p>
+   * If the HTTP statusCode is 300 or above a HttpException is throw
+   * which contains the HttpResponse. This is the cause in the
+   * CompletionException.
+   *
+   * <pre>{@code
+   *
+   *   CompletableFuture<Stream<Customer>> future = clientContext.request()
+   *       .path("customers/stream")
+   *       .GET().async()
+   *       .stream(Customer.class);
+   *
+   *   future.whenComplete((stream, throwable) -> {
+   *       // if throwable != null ... handle error
+   *
+   *       // else process Stream<Customer> ...
+   *       try (stream) {
+   *         stream.forEach(customer -> {
+   *           ...
+   *         });
+   *       }
+   *     });
+   *
+   * }</pre>
+   *
+   * @param type The bean type to convert the content to
+   * @return The CompletableFuture of the response
+   */
+  <E> CompletableFuture<Stream<E>> stream(Class<E> type);
 }

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

@@ -41,7 +41,37 @@ void get_stream() {
   }
 
   @Test
-  void async_stream() throws ExecutionException, InterruptedException {
+  void async_get_stream() throws ExecutionException, InterruptedException {
+
+    final CompletableFuture<Stream<SimpleData>> future = clientContext.request()
+      .path("hello").path("stream")
+      .GET().async()
+      .stream(SimpleData.class);
+
+    future.whenComplete((stream, throwable) -> {
+      assertThat(throwable).isNull();
+
+      final List<SimpleData> data = stream.collect(Collectors.toList());
+      assertThat(data).hasSize(4);
+      final SimpleData first = data.get(0);
+      assertThat(first.id).isEqualTo(1);
+      assertThat(first.name).isEqualTo("one");
+
+      try (stream) {
+        // more typically process with forEach ...
+        stream.forEach(simpleData -> {
+          System.out.println("process " + simpleData.id + " " + simpleData.name);
+        });
+      }
+    });
+
+    // wait ...
+    future.get();
+
+  }
+
+  @Test
+  void async_stream_fromLineSubscriber() throws ExecutionException, InterruptedException {
 
     AtomicReference<HttpResponse<Void>> hresRef = new AtomicReference<>();
     AtomicReference<Throwable> errRef = new AtomicReference<>();
@@ -59,14 +89,17 @@ 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);
@@ -274,11 +307,11 @@ void async_exceptionally_style() {
       .bean(HelloDto.class);
 
     future.exceptionally(throwable -> {
-        final HttpException httpException = (HttpException) throwable.getCause();
-        causeRef.set(httpException);
-        assertThat(httpException.getStatusCode()).isEqualTo(422);
+      final HttpException httpException = (HttpException) throwable.getCause();
+      causeRef.set(httpException);
+      assertThat(httpException.getStatusCode()).isEqualTo(422);
 
-        return new HelloDto(0, "ErrorResponse", "");
+      return new HelloDto(0, "ErrorResponse", "");
 
     }).thenAccept(helloDto -> {
       assertThat(helloDto.name).isEqualTo("ErrorResponse");