Support non-SharedFetch usage of HC Client

Author: itzg

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

@@ -29,7 +29,6 @@
 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;
@@ -189,10 +188,6 @@ 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());
@@ -204,8 +199,23 @@ protected <R> R useReactiveClient(ReactiveClientUser<R> user) {
         }
     }
 
-    protected CloseableHttpAsyncClient getHcAsyncClient() {
-        return state.sharedFetch.getHcAsyncClient();
+    @FunctionalInterface
+    protected interface HcAsyncClientUser<R> {
+        void use(CloseableHttpAsyncClient client, Runnable closer);
+    }
+
+    protected <R> void useHcAsyncClient(HcAsyncClientUser<R> user) {
+        if (state.sharedFetch != null) {
+            user.use(state.sharedFetch.getHcAsyncClient(), () -> {
+                }
+            );
+        }
+        else {
+            //noinspection resource since close callback will handle it
+            SharedFetch sharedFetch = new SharedFetch(state.userAgentCommand, Options.builder().build());
+
+            user.use(sharedFetch.getHcAsyncClient(), sharedFetch::close);
+        }
     }
 
     protected static BiConsumer<? super HttpClientRequest, ? super Connection> debugLogRequest(

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

@@ -219,11 +219,15 @@ else if (skipUpToDate) {
                         .map(instant -> reqBuilder.setHeader("If-Modified-Since", httpDateTimeFormatter.format(instant)))
                         .then(
                             Mono.<Path>create(sink -> {
-                                getHcAsyncClient().execute(
-                                    SimpleRequestProducer.create(reqBuilder.build()),
-                                    new ResponseToFileConsumer(outputFile),
-                                    new MonoSinkFutureCallbackAdapter<>(sink)
-                                );
+                                useHcAsyncClient((hcClient, close) -> {
+                                    sink.onDispose(close::run);
+
+                                    hcClient.execute(
+                                        SimpleRequestProducer.create(reqBuilder.build()),
+                                        new ResponseToFileConsumer(outputFile),
+                                        new MonoSinkFutureCallbackAdapter<>(sink)
+                                    );
+                                });
                             })
                         );
             })

src/main/java/me/itzg/helpers/sync/MulitCopyCommand.java

@@ -20,6 +20,7 @@
 import me.itzg.helpers.http.FailedRequestException;
 import me.itzg.helpers.http.Fetch;
 import me.itzg.helpers.http.SharedFetch;
+import me.itzg.helpers.http.SharedFetch.Options;
 import me.itzg.helpers.http.Uris;
 import org.jetbrains.annotations.Blocking;
 import org.reactivestreams.Publisher;
@@ -82,13 +83,15 @@ public Integer call() throws Exception {
 
         Files.createDirectories(dest);
 
-        Flux.fromIterable(sources)
-            .map(String::trim)
-            .filter(s -> !s.isEmpty())
-            .flatMap(source -> processSource(source, fileIsListingOption))
-            .collectList()
-            .flatMap(this::cleanupAndSaveManifest)
-            .block();
+        try (SharedFetch sharedFetch = Fetch.sharedFetch("mcopy", Options.builder().build())) {
+            Flux.fromIterable(sources)
+                .map(String::trim)
+                .filter(s -> !s.isEmpty())
+                .flatMap(source -> processSource(sharedFetch, source, fileIsListingOption))
+                .collectList()
+                .flatMap(this::cleanupAndSaveManifest)
+                .block();
+        }
 
         return ExitCode.OK;
     }
@@ -113,9 +116,9 @@ private Mono<?> cleanupAndSaveManifest(List<Path> paths) {
             });
     }
 
-    private Publisher<Path> processSource(String source, boolean fileIsListing) {
+    private Publisher<Path> processSource(SharedFetch sharedFetch, String source, boolean fileIsListing) {
         if (Uris.isUri(source)) {
-            return fileIsListing ? processRemoteListingFile(source) : processRemoteSource(source);
+            return fileIsListing ? processRemoteListingFile(source) : processRemoteSource(sharedFetch, source);
         } else {
             final Path path = Paths.get(source);
             if (!Files.exists(path)) {
@@ -125,12 +128,12 @@ private Publisher<Path> processSource(String source, boolean fileIsListing) {
             if (Files.isDirectory(path)) {
                 return processDirectory(path);
             } else {
-                return fileIsListing ? processListingFile(path) : processFile(path);
+                return fileIsListing ? processListingFile(sharedFetch, path) : processFile(path);
             }
         }
     }
 
-    private Flux<Path> processListingFile(Path listingFile) {
+    private Flux<Path> processListingFile(SharedFetch sharedFetch, Path listingFile) {
         return Mono.just(listingFile)
             .publishOn(Schedulers.boundedElastic())
             .flatMapMany(path -> {
@@ -139,7 +142,7 @@ private Flux<Path> processListingFile(Path listingFile) {
                     final List<String> lines = Files.readAllLines(path);
                     return Flux.fromIterable(lines)
                         .filter(this::isListingLine)
-                        .flatMap(src -> processSource(src,
+                        .flatMap(src -> processSource(sharedFetch, src,
                             // avoid recursive file-listing processing
                             false));
                 } catch (IOException e) {
@@ -230,8 +233,8 @@ private Flux<Path> processDirectory(Path srcDir) {
             });
     }
 
-    private Mono<Path> processRemoteSource(String source) {
-        return Fetch.fetch(URI.create(source))
+    private Mono<Path> processRemoteSource(SharedFetch sharedFetch, String source) {
+        return sharedFetch.fetch(URI.create(source))
             .userAgentCommand("mcopy")
             .toDirectory(dest)
             .skipUpToDate(skipUpToDate)
@@ -273,7 +276,7 @@ private Flux<Path> processRemoteListingFile(String source) {
                     .flatMapMany(content -> Flux.just(content.split("\\r?\\n")))
                     .filter(this::isListingLine)
             )
-            .flatMap(this::processRemoteSource)
+            .flatMap(sourceInListing -> processRemoteSource(sharedFetch, sourceInListing))
             .doOnTerminate(sharedFetch::close)
             .checkpoint("Processing remote listing at " + source, true);
     }