#17 - Add stream() ... for streaming beans from x-json-stream response

Author: rbygrave

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

@@ -10,4 +10,8 @@ public interface BodyReader<T> {
    */
   T read(BodyContent content);
 
+  /**
+   * Read the String content returning it as a java type.
+   */
+  T readBody(String content);
 }

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

@@ -159,6 +159,10 @@ BodyContent write(Object bean, String contentType) {
     return bodyAdapter.beanWriter(bean.getClass()).write(bean, contentType);
   }
 
+  <T> BodyReader<T> beanReader(Class<T> cls) {
+    return bodyAdapter.beanReader(cls);
+  }
+
   <T> T readBean(Class<T> cls, BodyContent content) {
     return bodyAdapter.beanReader(cls).read(content);
   }

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

@@ -348,6 +348,17 @@ public <T> List<T> list(Class<T> cls) {
     return context.readList(cls, encodedResponseBody);
   }
 
+  @Override
+  public <T> Stream<T> stream(Class<T> cls) {
+    final HttpResponse<Stream<String>> res = withResponseHandler(HttpResponse.BodyHandlers.ofLines());
+    this.httpResponse = res;
+    if (res.statusCode() >= 300) {
+      throw new HttpException(res, context);
+    }
+    final BodyReader<T> bodyReader = context.beanReader(cls);
+    return res.body().map(bodyReader::readBody);
+  }
+
   @Override
   public <T> HttpResponse<T> withResponseHandler(HttpResponse.BodyHandler<T> responseHandler) {
     context.beforeRequest(this);

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

@@ -42,6 +42,19 @@ public interface HttpClientResponse {
    */
   <T> List<T> list(Class<T> type);
 
+  /**
+   * Return the response as a stream of beans.
+   * <p>
+   * Typically the response is expected to be {@literal applciation/x-json-stream}
+   * newline delimited json payload.
+   *
+   * @param type The type of the bean to convert the response content into.
+   * @param <T>  The type that the content is converted to.
+   * @return The stream of beans from the response
+   * @throws HttpException when the response has error status codes
+   */
+  <T> Stream<T> stream(Class<T> type);
+
   /**
    * Return the response with check for 200 range status code.
    * <p>

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

@@ -83,6 +83,15 @@ private static class JReader<T> implements BodyReader<T> {
       this.reader = reader;
     }
 
+    @Override
+    public T readBody(String content) {
+      try {
+        return reader.readValue(content);
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
     @Override
     public T read(BodyContent s) {
       try {

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

@@ -7,6 +7,8 @@
 import java.net.http.HttpResponse;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.*;
@@ -15,6 +17,22 @@ class HelloControllerTest extends BaseWebTest {
 
   final HttpClientContext clientContext = client();
 
+  @Test
+  void get_stream() {
+
+    final Stream<SimpleData> stream = clientContext.request()
+      .path("hello").path("stream")
+      .GET()
+      .stream(SimpleData.class);
+
+    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");
+  }
+
   @Test
   void get_helloMessage() {
 
@@ -233,4 +251,9 @@ void get_withMatrixParam() {
     assertEquals(200, httpRes.statusCode());
     assertEquals("yr:2011 au:rob co:nz other:foo extra:banana", httpRes.body());
   }
+
+  public static class SimpleData {
+    public long id;
+    public String name;
+  }
 }

client/src/test/java/org/example/webserver/HelloController$Route.java

@@ -36,6 +36,11 @@ public void registerRoutes() {
       ctx.contentType("text/plain").result(controller.retry());
     });
 
+    ApiBuilder.get("/hello/stream", ctx -> {
+      ctx.status(200);
+      controller.stream(ctx);
+    });
+
     ApiBuilder.get("/hello/:id/:date", ctx -> {
       ctx.status(200);
       int id = asInt(ctx.pathParam("id"));

client/src/test/java/org/example/webserver/HelloController.java

@@ -47,6 +47,18 @@ String retry() {
     }
   }
 
+  @Get("stream")
+  void stream(Context context) {
+    // simulate x-json-stream response
+    context.header("content-type", "application/x-json-stream");
+    String content =
+      "{\"id\":1, \"name\":\"one\"}\n" +
+        "{\"id\":2, \"name\":\"two\"}\n" +
+        "{\"id\":3, \"name\":\"three\"}\n" +
+        "{\"id\":4, \"name\":\"four\"}\n";
+    context.result(content);
+  }
+
   /**
    * Return the Hello DTO.
    *

gson-adapter/pom.xml

@@ -27,7 +27,7 @@
     <dependency>
       <groupId>io.avaje</groupId>
       <artifactId>avaje-http-client</artifactId>
-      <version>1.3</version>
+      <version>1.6-SNAPSHOT</version>
       <scope>provided</scope>
     </dependency>
 

gson-adapter/src/main/java/io/avaje/http/client/gson/GsonBodyAdapter.java

@@ -100,6 +100,18 @@ private static class Reader<T> implements BodyReader<T> {
       this.adapter = adapter;
     }
 
+    /**
+     * Read the content returning it as a java type.
+     */
+    @Override
+    public T readBody(String content) {
+      try {
+        return adapter.fromJson(content);
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
     @Override
     public T read(BodyContent body) {
       try {