Re-use HttpAsyncClient

Author: itzg

src/main/java/me/itzg/helpers/http/FetchBuilderBase.java

@@ -29,7 +29,9 @@
 import me.itzg.helpers.errors.GenericException;
 import me.itzg.helpers.http.SharedFetch.Options;
 import me.itzg.helpers.json.ObjectMappers;
+import org.apache.hc.client5.http.async.HttpAsyncClient;
 import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
 import org.slf4j.Logger;
 import reactor.core.publisher.Mono;
 import reactor.netty.ByteBufMono;
@@ -177,6 +179,10 @@ protected interface ReactiveClientUser<R> {
         R use(HttpClient client);
     }
 
+    protected interface HcAsyncClientUser<R> {
+        R use(HttpAsyncClient client);
+    }
+
     protected <R> R useReactiveClient(ReactiveClientUser<R> user) {
         if (state.sharedFetch != null) {
             return user.use(state.sharedFetch.getReactiveClient());
@@ -188,6 +194,10 @@ protected <R> R useReactiveClient(ReactiveClientUser<R> user) {
         }
     }
 
+    protected CloseableHttpAsyncClient getHcAsyncClient() {
+        return state.sharedFetch.getHcAsyncClient();
+    }
+
     protected static BiConsumer<? super HttpClientRequest, ? super Connection> debugLogRequest(
         Logger log, String operation
     ) {

src/main/java/me/itzg/helpers/http/OutputToDirectoryFetchBuilder.java

@@ -18,8 +18,6 @@
 import org.apache.hc.client5.http.async.methods.AbstractBinResponseConsumer;
 import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
 import org.apache.hc.client5.http.async.methods.SimpleRequestProducer;
-import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
-import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.HttpException;
 import org.apache.hc.core5.http.HttpResponse;
@@ -36,7 +34,6 @@
 public class OutputToDirectoryFetchBuilder extends FetchBuilderBase<OutputToDirectoryFetchBuilder> {
 
     private final Path outputDirectory;
-    private final CloseableHttpAsyncClient hcHttpClient;
 
     @Setter
     private boolean skipExisting;
@@ -56,9 +53,6 @@ protected OutputToDirectoryFetchBuilder(State state, Path outputDirectory) {
             throw new IllegalArgumentException(outputDirectory + " is not a directory or does not exist");
         }
         this.outputDirectory = outputDirectory;
-
-        hcHttpClient = HttpAsyncClients.createDefault();
-        hcHttpClient.start();
     }
 
     @SuppressWarnings("unused")
@@ -225,7 +219,7 @@ else if (skipUpToDate) {
                         .map(instant -> reqBuilder.setHeader("If-Modified-Since", httpDateTimeFormatter.format(instant)))
                         .then(
                             Mono.<Path>create(sink -> {
-                                hcHttpClient.execute(
+                                getHcAsyncClient().execute(
                                     SimpleRequestProducer.create(reqBuilder.build()),
                                     new ResponseToFileConsumer(outputFile),
                                     new MonoSinkFutureCallbackAdapter<>(sink)

src/main/java/me/itzg/helpers/http/SharedFetch.java

@@ -1,6 +1,7 @@
 package me.itzg.helpers.http;
 
 import io.netty.handler.codec.http.HttpHeaderNames;
+import java.io.IOException;
 import java.net.URI;
 import java.time.Duration;
 import java.util.HashMap;
@@ -12,6 +13,8 @@
 import lombok.extern.slf4j.Slf4j;
 import me.itzg.helpers.McImageHelper;
 import me.itzg.helpers.errors.GenericException;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
 import reactor.netty.http.Http11SslContextSpec;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.resources.ConnectionProvider;
@@ -32,6 +35,8 @@ public class SharedFetch implements AutoCloseable {
     final LatchingUrisInterceptor latchingUrisInterceptor = new LatchingUrisInterceptor();
 
     private final HttpClient reactiveClient;
+    private final CloseableHttpAsyncClient hcAsyncClient;
+    private boolean hcAsyncClientStarted = false;
 
     public SharedFetch(String forCommand, Options options) {
         final String userAgent = String.format("%s/%s/%s (cmd=%s)",
@@ -74,6 +79,16 @@ public SharedFetch(String forCommand, Options options) {
             );
 
         headers.put("x-fetch-session", fetchSessionId);
+
+        hcAsyncClient = HttpAsyncClients.createSystem();
+    }
+
+    public synchronized CloseableHttpAsyncClient getHcAsyncClient() {
+        if (!hcAsyncClientStarted) {
+            hcAsyncClient.start();
+            hcAsyncClientStarted = true;
+        }
+        return hcAsyncClient;
     }
 
     public FetchBuilderBase<?> fetch(URI uri) {
@@ -88,6 +103,11 @@ public SharedFetch addHeader(String name, String value) {
 
     @Override
     public void close() {
+        try {
+            hcAsyncClient.close();
+        } catch (IOException e) {
+            log.warn("Failed to close async client for shared fetch", e);
+        }
     }
 
     @Builder