Merge pull request #18 from avaje/feature/async

Add support for async method calls

Author: rbygrave

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

@@ -0,0 +1,47 @@
+package io.avaje.http.client;
+
+import java.net.http.HttpResponse;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+class DHttpAsync implements HttpAsyncResponse {
+
+  private final DHttpClientRequest request;
+
+  DHttpAsync(DHttpClientRequest request) {
+    this.request = request;
+  }
+
+  @Override
+  public <E> CompletableFuture<HttpResponse<E>> withHandler(HttpResponse.BodyHandler<E>  handler) {
+    return request
+      .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
+      .performSendAsync(true, HttpResponse.BodyHandlers.ofString())
+      .thenApply(request::afterAsync);
+  }
+
+  @Override
+  public <E> CompletableFuture<E> bean(Class<E> type) {
+    return request
+      .performSendAsync(true, HttpResponse.BodyHandlers.ofByteArray())
+      .thenApply(httpResponse -> request.asyncBean(type, httpResponse));
+  }
+
+  @Override
+  public <E> CompletableFuture<List<E>> list(Class<E> type) {
+    return request
+      .performSendAsync(true, HttpResponse.BodyHandlers.ofByteArray())
+      .thenApply(httpResponse -> request.asyncList(type, httpResponse));
+  }
+}

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

@@ -9,6 +9,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 
 class DHttpClientContext implements HttpClientContext {
@@ -29,6 +30,7 @@ class DHttpClientContext implements HttpClientContext {
   private final boolean withAuthToken;
   private final AuthTokenProvider authTokenProvider;
   private final AtomicReference<AuthToken> tokenRef = new AtomicReference<>();
+  private int loggingMaxBody = 1_000;
 
   DHttpClientContext(HttpClient httpClient, String baseUrl, Duration requestTimeout, BodyAdapter bodyAdapter, RetryHandler retryHandler, RequestListener requestListener, AuthTokenProvider authTokenProvider, RequestIntercept intercept) {
     this.httpClient = httpClient;
@@ -98,7 +100,7 @@ public void checkResponse(HttpResponse<?> response) {
     }
   }
 
-  void check(HttpResponse<byte[]> response) {
+  void checkMaybeThrow(HttpResponse<byte[]> response) {
     if (response.statusCode() >= 300) {
       throw new HttpException(this, response);
     }
@@ -155,6 +157,10 @@ <T> HttpResponse<T> send(HttpRequest.Builder requestBuilder, HttpResponse.BodyHa
     }
   }
 
+  <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest.Builder requestBuilder, HttpResponse.BodyHandler<T> bodyHandler) {
+    return httpClient.sendAsync(requestBuilder.build(), bodyHandler);
+  }
+
   BodyContent write(Object bean, String contentType) {
     return bodyAdapter.beanWriter(bean.getClass()).write(bean, contentType);
   }
@@ -198,4 +204,7 @@ private String authToken() {
     return authToken.token();
   }
 
+  String maxResponseBody(String body) {
+    return body.length() > loggingMaxBody ? body.substring(0, loggingMaxBody) + " <truncated> ..." : body;
+  }
 }

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

@@ -13,6 +13,7 @@
 import java.nio.file.Path;
 import java.time.*;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
@@ -49,6 +50,7 @@ class DHttpClientRequest implements HttpClientRequest, HttpClientResponse {
   private boolean loggableResponseBody;
   private boolean skipAuthToken;
   private boolean suppressLogging;
+  private long startAsyncNanos;
 
   DHttpClientRequest(DHttpClientContext context, Duration requestTimeout) {
     this.context = context;
@@ -276,6 +278,11 @@ private void addHeaders() {
     }
   }
 
+  @Override
+  public HttpAsyncResponse async() {
+    return new DHttpAsync(this);
+  }
+
   @Override
   public HttpClientResponse HEAD() {
     httpRequest = newHead(url.build());
@@ -320,7 +327,7 @@ public HttpClientResponse TRACE() {
   private void readResponseContent() {
     final HttpResponse<byte[]> response = asByteArray();
     this.httpResponse = response;
-    context.check(response);
+    context.checkMaybeThrow(response);
     encodedResponseBody = context.readContent(response);
   }
 
@@ -350,7 +357,7 @@ public <T> List<T> list(Class<T> cls) {
 
   @Override
   public <T> Stream<T> stream(Class<T> cls) {
-    final HttpResponse<Stream<String>> res = withResponseHandler(HttpResponse.BodyHandlers.ofLines());
+    final HttpResponse<Stream<String>> res = withHandler(HttpResponse.BodyHandlers.ofLines());
     this.httpResponse = res;
     if (res.statusCode() >= 300) {
       throw new HttpException(res, context);
@@ -360,7 +367,7 @@ public <T> Stream<T> stream(Class<T> cls) {
   }
 
   @Override
-  public <T> HttpResponse<T> withResponseHandler(HttpResponse.BodyHandler<T> responseHandler) {
+  public <T> HttpResponse<T> withHandler(HttpResponse.BodyHandler<T> responseHandler) {
     context.beforeRequest(this);
     addHeaders();
     HttpResponse<T> response = performSend(responseHandler);
@@ -378,35 +385,68 @@ protected <T> HttpResponse<T> performSend(HttpResponse.BodyHandler<T> responseHa
     }
   }
 
+  protected <T> CompletableFuture<HttpResponse<T>> performSendAsync(boolean loggable, HttpResponse.BodyHandler<T> responseHandler) {
+    loggableResponseBody = loggable;
+    context.beforeRequest(this);
+    addHeaders();
+    startAsyncNanos = System.nanoTime();
+    return context.sendAsync(httpRequest, responseHandler);
+  }
+
+  protected <E> E asyncBean(Class<E> type, HttpResponse<byte[]> response) {
+    afterAsyncEncoded(response);
+    return context.readBean(type, encodedResponseBody);
+  }
+
+  protected <E> List<E> asyncList(Class<E> type, HttpResponse<byte[]> response) {
+    afterAsyncEncoded(response);
+    return context.readList(type, encodedResponseBody);
+  }
+
+  private void afterAsyncEncoded(HttpResponse<byte[]> response) {
+    requestTimeNanos = System.nanoTime() - startAsyncNanos;
+    httpResponse = response;
+    encodedResponseBody = context.readContent(response);
+    context.afterResponse(this);
+    context.checkMaybeThrow(response);
+  }
+
+  protected <E> HttpResponse<E> afterAsync(HttpResponse<E> response) {
+    requestTimeNanos = System.nanoTime() - startAsyncNanos;
+    httpResponse = response;
+    context.afterResponse(this);
+    return response;
+  }
+
   @Override
   public HttpResponse<byte[]> asByteArray() {
-    return withResponseHandler(HttpResponse.BodyHandlers.ofByteArray());
+    return withHandler(HttpResponse.BodyHandlers.ofByteArray());
   }
 
   @Override
   public HttpResponse<String> asString() {
     loggableResponseBody = true;
-    return withResponseHandler(HttpResponse.BodyHandlers.ofString());
+    return withHandler(HttpResponse.BodyHandlers.ofString());
   }
 
   @Override
   public HttpResponse<Void> asDiscarding() {
-    return withResponseHandler(discarding());
+    return withHandler(discarding());
   }
 
   @Override
   public HttpResponse<InputStream> asInputStream() {
-    return withResponseHandler(HttpResponse.BodyHandlers.ofInputStream());
+    return withHandler(HttpResponse.BodyHandlers.ofInputStream());
   }
 
   @Override
   public HttpResponse<Path> asFile(Path file) {
-    return withResponseHandler(HttpResponse.BodyHandlers.ofFile(file));
+    return withHandler(HttpResponse.BodyHandlers.ofFile(file));
   }
 
   @Override
   public HttpResponse<Stream<String>> asLines() {
-    return withResponseHandler(HttpResponse.BodyHandlers.ofLines());
+    return withHandler(HttpResponse.BodyHandlers.ofLines());
   }
 
   private HttpRequest.Builder newReq(String url) {
@@ -511,13 +551,10 @@ public String responseBody() {
         return "<suppressed response body>";
       }
       if (encodedResponseBody != null) {
-        return new String(encodedResponseBody.content(), StandardCharsets.UTF_8);
+        return context.maxResponseBody(new String(encodedResponseBody.content(), StandardCharsets.UTF_8));
       } else if (httpResponse != null && loggableResponseBody) {
-        String strBody = httpResponse.body().toString();
-        if (strBody.length() > 1_000) {
-          return strBody.substring(0, 1_000) + "...";
-        }
-        return strBody;
+        final Object body = httpResponse.body();
+        return (body == null) ? null : context.maxResponseBody(body.toString());
       }
       return null;
     }
@@ -527,7 +564,7 @@ static class HttpVoidResponse implements HttpResponse<Void> {
 
     private final HttpResponse<?> orig;
 
-    @SuppressWarnings({"unchecked", "raw"})
+    @SuppressWarnings({"raw"})
     HttpVoidResponse(HttpResponse<?> orig) {
       this.orig = orig;
     }