Merge branch 'master' of github.com:avaje/avaje-http-client

Author: rbygrave

client/pom.xml

@@ -34,7 +34,7 @@
     <dependency>
       <groupId>io.avaje</groupId>
       <artifactId>avaje-jsonb</artifactId>
-      <version>0.5</version>
+      <version>0.8</version>
       <optional>true</optional>
     </dependency>
 

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

@@ -11,7 +11,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
 
 class DHttpClientContext implements HttpClientContext {
 
@@ -31,6 +33,7 @@ class DHttpClientContext implements HttpClientContext {
   private final boolean withAuthToken;
   private final AuthTokenProvider authTokenProvider;
   private final AtomicReference<AuthToken> tokenRef = new AtomicReference<>();
+  private final AtomicLong activeAsync = new AtomicLong();
   private int loggingMaxBody = 1_000;
 
   DHttpClientContext(HttpClient httpClient, String baseUrl, Duration requestTimeout, BodyAdapter bodyAdapter, RetryHandler retryHandler, RequestListener requestListener, AuthTokenProvider authTokenProvider, RequestIntercept intercept) {
@@ -173,6 +176,7 @@ <T> HttpResponse<T> send(HttpRequest.Builder requestBuilder, HttpResponse.BodyHa
   }
 
   <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest.Builder requestBuilder, HttpResponse.BodyHandler<T> bodyHandler) {
+    activeAsync.incrementAndGet();
     return httpClient.sendAsync(requestBuilder.build(), bodyHandler);
   }
 
@@ -192,13 +196,28 @@ <T> List<T> readList(Class<T> cls, BodyContent content) {
     return bodyAdapter.listReader(cls).read(content);
   }
 
+  @Override
+  public boolean waitForAsync(long millis) {
+    final long until = System.currentTimeMillis() + millis;
+    do {
+      if (activeAsync.get() <= 0) {
+        return true;
+      }
+      LockSupport.parkNanos(10_000_000);
+    } while (System.currentTimeMillis() < until);
+    return false;
+  }
+
   void afterResponse(DHttpClientRequest request) {
     if (requestListener != null) {
       requestListener.response(request.listenerEvent());
     }
     if (requestIntercept != null) {
       requestIntercept.afterResponse(request.response(), request);
     }
+    if (request.startAsyncNanos > 0) {
+      activeAsync.decrementAndGet();
+    }
   }
 
   void beforeRequest(DHttpClientRequest request) {

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

@@ -50,7 +50,7 @@ class DHttpClientRequest implements HttpClientRequest, HttpClientResponse {
   private boolean loggableResponseBody;
   private boolean skipAuthToken;
   private boolean suppressLogging;
-  private long startAsyncNanos;
+  protected long startAsyncNanos;
   private String label;
   private Map<String, Object> customAttributes;
 

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

@@ -115,6 +115,14 @@ static HttpClientContext.Builder newBuilder() {
    */
   byte[] decodeContent(String encoding, byte[] content);
 
+  /**
+   * Wait for any submitted async requests with a given maximum wait time.
+   *
+   * @param maxWaitMillis The maximum time to wait in milliseconds
+   * @return True if waiting was successful or false if there are still async requests that have not yet come back for completion.
+   */
+  boolean waitForAsync(long maxWaitMillis);
+
   /**
    * Builds the HttpClientContext.
    *

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

@@ -0,0 +1,40 @@
+package io.avaje.http.client;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.http.HttpResponse;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class AsyncTest extends BaseWebTest {
+
+  final HttpClientContext clientContext = client();
+
+  @Test
+  void waitForAsync()  {
+    final CompletableFuture<HttpResponse<Stream<String>>> future = clientContext.request()
+      .path("hello").path("stream")
+      .GET()
+      .async()
+      .asLines();
+
+    final AtomicBoolean flag = new AtomicBoolean();
+    future.whenComplete((hres, throwable) -> {
+      flag.set(true);
+      assertThat(hres.statusCode()).isEqualTo(200);
+      List<String> lines = hres.body().collect(Collectors.toList());
+      assertThat(lines).hasSize(4);
+      assertThat(lines.get(0)).contains("{\"id\":1, \"name\":\"one\"}");
+    });
+
+    assertThat(flag).isFalse();
+    assertThat(clientContext.waitForAsync(1_000)).isTrue();
+    assertThat(flag).isTrue();
+  }
+
+}

client/src/test/java/org/example/github/GithubTest.java

@@ -13,32 +13,34 @@ public class GithubTest {
 
   @Test
   @Disabled
-  void test() throws InterruptedException {
+  void test() {
 
     final HttpClientContext clientContext = HttpClientContext.newBuilder()
       .baseUrl("https://api.github.com")
       .bodyAdapter(new JacksonBodyAdapter())
       .requestLogging(false)
       .build();
 
+    // will not work under module classpath without registering the HttpApiProvider
+    final Simple simple = clientContext.create(Simple.class);
+
+    final List<Repo> repos = simple.listRepos("rbygrave", "junk");
+    assertThat(repos).isNotEmpty();
+
     clientContext.request()
       .path("users").path("rbygrave").path("repos")
       .GET()
       .async()
       .asString()
       .thenAccept(res -> {
-
-        System.out.println("RES: "+res.statusCode());
-        System.out.println("BODY: "+res.body());
+        System.out.println("RES: " + res.statusCode());
+        System.out.println("BODY: " + res.body().substring(0, 150) + "...");
       });
 
-    Thread.sleep(1_000);
-
-    // will not work under module classpath without registering the HttpApiProvider
-    final Simple simple = clientContext.create(Simple.class);
-
-    final List<Repo> repos = simple.listRepos("rbygrave", "junk");
-    assertThat(repos).isNotEmpty();
+    long st = System.currentTimeMillis();
+    System.out.println("waitForAsync");
+    boolean waitSuccess = clientContext.waitForAsync(2_000);
+    System.out.println("waitForAsync waitSuccess:" + waitSuccess + " waitMillis: " + (System.currentTimeMillis() - st));
   }
 
 }