Parallelize artifact transfers with live console progress

src/main/java/rife/bld/dependencies/ArtifactRetriever.java

@@ -19,6 +19,7 @@ import java.net.URLConnection;
 import java.nio.channels.Channels;
 import java.security.MessageDigest;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 
 import static rife.tools.HttpUtils.HEADER_AUTHORIZATION;
 import static rife.tools.HttpUtils.basicAuthorizationHeader;
@@ -35,6 +36,8 @@ import static rife.tools.StringUtils.encodeHexLower;
  * @since 1.5.18
  */
 public abstract class ArtifactRetriever {
+    private static final int TRANSFER_CHUNK_SIZE = 128 * 1024;
+
     private final static ArtifactRetriever UNCACHED = new ArtifactRetriever() {
         String getCached(RepositoryArtifact artifact) {
             return null;
@@ -63,7 +66,7 @@ public abstract class ArtifactRetriever {
      */
     public static ArtifactRetriever cachingInstance() {
         return new ArtifactRetriever() {
-            private final Map<RepositoryArtifact, String> artifactCache = new HashMap<>();
+            private final Map<RepositoryArtifact, String> artifactCache = new ConcurrentHashMap<>();
 
             String getCached(RepositoryArtifact artifact) {
                 return artifactCache.get(artifact);
@@ -141,17 +144,17 @@ public abstract class ArtifactRetriever {
 
         var download_filename = artifact.location().substring(artifact.location().lastIndexOf("/") + 1);
         var download_file = new File(directory, download_filename);
-        System.out.print("Downloading: " + artifact.location() + " ... ");
-        System.out.flush();
+        var transfer = TransferOutput.instance().start(artifact.location());
+        var status = "";
         try {
             if (artifact.repository().isLocal()) {
                 var source = new File(artifact.location());
                 if (source.exists()) {
                     FileUtils.copy(source, download_file);
-                    System.out.print("done");
+                    status = "done";
                     return true;
                 } else {
-                    System.out.print("not found");
+                    status = "not found";
                     return false;
                 }
             } else {
@@ -159,7 +162,7 @@ public abstract class ArtifactRetriever {
                     if (download_file.exists() && download_file.canRead()) {
                         if (checkHash(artifact, download_file, ".sha256", "SHA-256") ||
                             checkHash(artifact, download_file, ".md5", "MD5")) {
-                            System.out.print("exists");
+                            status = "exists";
                             return true;
                         }
                     }
@@ -170,23 +173,29 @@ public abstract class ArtifactRetriever {
                             HEADER_AUTHORIZATION,
                             basicAuthorizationHeader(artifact.repository().username(), artifact.repository().password()));
                     }
+                    var content_length = connection.getContentLengthLong();
                     try (var input_stream = connection.getInputStream()) {
                         var readableByteChannel = Channels.newChannel(input_stream);
                         try (var fileOutputStream = new FileOutputStream(download_file)) {
                             var fileChannel = fileOutputStream.getChannel();
-                            fileChannel.transferFrom(readableByteChannel, 0, Long.MAX_VALUE);
+                            var position = 0L;
+                            long transferred;
+                            while ((transferred = fileChannel.transferFrom(readableByteChannel, position, TRANSFER_CHUNK_SIZE)) > 0) {
+                                position += transferred;
+                                transfer.progress(position, content_length);
+                            }
 
-                            System.out.print("done");
+                            status = "done";
                             return true;
                         }
                     }
                 } catch (FileNotFoundException e) {
-                    System.out.print("not found");
+                    status = "not found";
                     return false;
                 }
             }
         } finally {
-            System.out.println();
+            transfer.finish(status);
         }
     }
 

src/main/java/rife/bld/dependencies/DependencySet.java

@@ -8,6 +8,10 @@ import rife.bld.dependencies.exceptions.DependencyTransferException;
 
 import java.io.File;
 import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.Supplier;
 
 /**
  * Convenience class to handle a set of {@link Dependency} objects.
@@ -128,6 +132,11 @@ public class DependencySet extends AbstractSet<Dependency> implements Set<Depend
      * including other classifiers.
      * <p>
      * The destination directory must exist and be writable.
+     * <p>
+     * The artifacts of different dependencies are transferred in parallel,
+     * the {@value VersionResolution#PROPERTY_TRANSFER_PARALLELISM} property
+     * can be used to change the number of simultaneous transfers, setting it
+     * to {@code 1} makes the transfers sequential.
      *
      * @param resolution        the version resolution state that can be cached
      * @param retriever         the retriever to use to get artifacts
@@ -140,7 +149,7 @@ public class DependencySet extends AbstractSet<Dependency> implements Set<Depend
      * @since 2.1
      */
     public List<RepositoryArtifact> transferIntoDirectory(VersionResolution resolution, ArtifactRetriever retriever, List<Repository> repositories, File directory, File modulesDirectory, String... classifiers) {
-        var result = new ArrayList<RepositoryArtifact>();
+        var transfers = new ArrayList<Supplier<List<RepositoryArtifact>>>();
         for (var dependency : this) {
             var transfer_directory = directory;
             if (dependency.isModularJar()) {
@@ -159,21 +168,66 @@ public class DependencySet extends AbstractSet<Dependency> implements Set<Depend
                 }
             }
 
-            var artifact = new DependencyResolver(resolution, retriever, repositories, dependency).transferIntoDirectory(transfer_directory);
-            if (artifact != null) {
-                result.add(artifact);
-            }
+            final var target_directory = transfer_directory;
+            transfers.add(() -> {
+                var artifacts = new ArrayList<RepositoryArtifact>();
+                var artifact = new DependencyResolver(resolution, retriever, repositories, dependency).transferIntoDirectory(target_directory);
+                if (artifact != null) {
+                    artifacts.add(artifact);
+                }
 
-            if (classifiers != null) {
-                for (var classifier : classifiers) {
-                    if (classifier != null && !dependency.excludedClassifiers().contains(classifier)) {
-                        var classifier_artifact = new DependencyResolver(resolution, retriever, repositories, dependency.withClassifier(classifier)).transferIntoDirectory(transfer_directory);
-                        if (classifier_artifact != null) {
-                            result.add(classifier_artifact);
+                if (classifiers != null) {
+                    for (var classifier : classifiers) {
+                        if (classifier != null && !dependency.excludedClassifiers().contains(classifier)) {
+                            var classifier_artifact = new DependencyResolver(resolution, retriever, repositories, dependency.withClassifier(classifier)).transferIntoDirectory(target_directory);
+                            if (classifier_artifact != null) {
+                                artifacts.add(classifier_artifact);
+                            }
                         }
                     }
                 }
+                return artifacts;
+            });
+        }
+
+        return executeTransfers(transfers, resolution.transferParallelism());
+    }
+
+    private static List<RepositoryArtifact> executeTransfers(List<Supplier<List<RepositoryArtifact>>> transfers, int transferParallelism) {
+        var result = new ArrayList<RepositoryArtifact>();
+
+        var parallelism = Math.min(transfers.size(), transferParallelism);
+        if (parallelism <= 1) {
+            for (var transfer : transfers) {
+                result.addAll(transfer.get());
             }
+            return result;
+        }
+
+        var executor = Executors.newFixedThreadPool(parallelism);
+        try {
+            var futures = new ArrayList<Future<List<RepositoryArtifact>>>();
+            for (var transfer : transfers) {
+                futures.add(executor.submit(transfer::get));
+            }
+            for (var future : futures) {
+                try {
+                    result.addAll(future.get());
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new IllegalStateException("Artifact transfer was interrupted", e);
+                } catch (ExecutionException e) {
+                    if (e.getCause() instanceof RuntimeException runtime) {
+                        throw runtime;
+                    }
+                    if (e.getCause() instanceof Error error) {
+                        throw error;
+                    }
+                    throw new IllegalStateException(e.getCause());
+                }
+            }
+        } finally {
+            executor.shutdownNow();
         }
         return result;
     }

src/main/java/rife/bld/dependencies/TransferOutput.java

@@ -0,0 +1,185 @@
+/*
+ * Copyright 2001-2026 Geert Bevin (gbevin[remove] at uwyn dot com)
+ * Licensed under the Apache License, Version 2.0 (the "License")
+ */
+package rife.bld.dependencies;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Outputs the status of artifact transfers to the console.
+ * <p>
+ * When the console is an interactive terminal, the transfers that are in
+ * progress are displayed in a live updating block at the bottom of the
+ * output, with each transfer reporting its progress as it happens.
+ * Otherwise, a single line is printed when each transfer finishes, keeping
+ * the output stable for CI logs, pipes and IDE consoles, while still never
+ * interleaving the lines of parallel transfers.
+ *
+ * @author Geert Bevin (gbevin[remove] at uwyn dot com)
+ * @since 2.3.1
+ */
+abstract class TransferOutput {
+    private static final TransferOutput INSTANCE = create();
+
+    static TransferOutput instance() {
+        return INSTANCE;
+    }
+
+    private static TransferOutput create() {
+        if (System.console() != null && !"dumb".equals(System.getenv("TERM"))) {
+            return new AnsiTransferOutput();
+        }
+        return new PlainTransferOutput();
+    }
+
+    /**
+     * Starts tracking a single artifact transfer.
+     *
+     * @param location the location of the artifact that is being transferred
+     * @return the transfer to report progress and completion on
+     * @since 2.3.1
+     */
+    abstract Transfer start(String location);
+
+    /**
+     * A single artifact transfer whose progress is being tracked.
+     *
+     * @since 2.3.1
+     */
+    abstract static class Transfer {
+        protected final String location_;
+
+        protected Transfer(String location) {
+            location_ = location;
+        }
+
+        /**
+         * Reports transfer progress.
+         *
+         * @param transferred the number of bytes transferred so far
+         * @param total       the total number of bytes; or {@code -1} when unknown
+         * @since 2.3.1
+         */
+        abstract void progress(long transferred, long total);
+
+        /**
+         * Reports that the transfer finished.
+         *
+         * @param status the final status of the transfer, like
+         *               {@code done}, {@code exists} or {@code not found}
+         * @since 2.3.1
+         */
+        abstract void finish(String status);
+    }
+
+    static String describe(String location, String status) {
+        return "Downloading: " + location + " ... " + status;
+    }
+
+    private static class PlainTransferOutput extends TransferOutput {
+        Transfer start(String location) {
+            return new Transfer(location) {
+                void progress(long transferred, long total) {
+                    // no-op, only the finished state is printed
+                }
+
+                void finish(String status) {
+                    System.out.println(describe(location_, status));
+                }
+            };
+        }
+    }
+
+    private static class AnsiTransferOutput extends TransferOutput {
+        private static final long REPAINT_INTERVAL_MS = 100;
+        private static final int MAX_LINE_WIDTH = 79;
+
+        private final List<AnsiTransfer> active_ = new ArrayList<>();
+        private int paintedLines_ = 0;
+        private long lastRepaint_ = 0;
+
+        synchronized Transfer start(String location) {
+            var transfer = new AnsiTransfer(location);
+            active_.add(transfer);
+            repaint(null);
+            return transfer;
+        }
+
+        private synchronized void reportProgress() {
+            if (System.currentTimeMillis() - lastRepaint_ >= REPAINT_INTERVAL_MS) {
+                repaint(null);
+            }
+        }
+
+        private synchronized void reportFinished(AnsiTransfer transfer, String status) {
+            active_.remove(transfer);
+            repaint(describe(transfer.location_, status));
+        }
+
+        private void repaint(String finishedLine) {
+            var output = new StringBuilder();
+            if (paintedLines_ > 0) {
+                // move the cursor to the beginning of the live block
+                // and clear everything below it
+                output.append("\u001B[").append(paintedLines_).append("F\u001B[0J");
+            }
+            if (finishedLine != null) {
+                output.append(finishedLine).append('\n');
+            }
+            for (var transfer : active_) {
+                output.append(transfer.statusLine()).append('\n');
+            }
+            paintedLines_ = active_.size();
+            lastRepaint_ = System.currentTimeMillis();
+            System.out.print(output);
+            System.out.flush();
+        }
+
+        private class AnsiTransfer extends Transfer {
+            private volatile long transferred_ = 0;
+            private volatile long total_ = -1;
+
+            AnsiTransfer(String location) {
+                super(location);
+            }
+
+            void progress(long transferred, long total) {
+                transferred_ = transferred;
+                total_ = total;
+                reportProgress();
+            }
+
+            void finish(String status) {
+                reportFinished(this, status);
+            }
+
+            String statusLine() {
+                var filename = location_.substring(location_.lastIndexOf('/') + 1);
+                var line = new StringBuilder("Downloading: ").append(filename).append(" ... ");
+                if (total_ > 0) {
+                    line.append(transferred_ * 100 / total_).append('%');
+                } else if (transferred_ > 0) {
+                    line.append(humanBytes(transferred_));
+                }
+                // prevent lines from wrapping since that would break
+                // the cursor repositioning of the repaints
+                if (line.length() > MAX_LINE_WIDTH) {
+                    line.setLength(MAX_LINE_WIDTH);
+                }
+                return line.toString();
+            }
+
+            private String humanBytes(long bytes) {
+                if (bytes >= 1024 * 1024) {
+                    return (bytes / (1024 * 1024)) + " MB";
+                }
+                if (bytes >= 1024) {
+                    return (bytes / 1024) + " KB";
+                }
+                return bytes + " B";
+            }
+        }
+    }
+}

src/main/java/rife/bld/dependencies/VersionResolution.java

@@ -8,6 +8,7 @@ import rife.ioc.HierarchicalProperties;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.logging.Logger;
 
 /**
  * This class is responsible for managing version overrides for dependencies.
@@ -27,6 +28,10 @@ import java.util.Map;
  * bld.override-tests=com.uwyn.rife2:bld-tests-badge:1.4.7
  * bld.override-h2=com.h2database:h2:2.2.222
  * </pre>
+ * <p>
+ * It also captures other dependency resolution preferences, like the number
+ * of parallel artifact transfers through the "{@code bld.transferParallelism}"
+ * property.
  * @since 2.0
  */
 public class VersionResolution {
@@ -36,7 +41,16 @@ public class VersionResolution {
      */
     public static final String PROPERTY_OVERRIDE_PREFIX = "bld.override";
 
+    /**
+     * The property key that determines how many artifact transfers are
+     * performed in parallel, {@code 1} makes them sequential.
+     * @since 2.3.1
+     */
+    public static final String PROPERTY_TRANSFER_PARALLELISM = "bld.transferParallelism";
+    private static final int DEFAULT_TRANSFER_PARALLELISM = 6;
+
     private final Map<String, Version> versionOverrides_ = new HashMap<>();
+    private final int transferParallelism_;
 
     /**
      * Returns a dummy {@code VersionResolution} instance that doesn't override anything.
@@ -59,6 +73,7 @@ public class VersionResolution {
      * @since 2.0
      */
     public VersionResolution(HierarchicalProperties properties) {
+        var transfer_parallelism = DEFAULT_TRANSFER_PARALLELISM;
         if (properties != null) {
             for (var name : properties.getNames()) {
                 if (name.startsWith(PROPERTY_OVERRIDE_PREFIX)) {
@@ -73,7 +88,17 @@ public class VersionResolution {
                     }
                 }
             }
+
+            var parallelism = properties.getValueString(PROPERTY_TRANSFER_PARALLELISM);
+            if (parallelism != null && !parallelism.isBlank()) {
+                try {
+                    transfer_parallelism = Math.max(1, Integer.parseInt(parallelism.trim()));
+                } catch (NumberFormatException e) {
+                    Logger.getLogger("rife.bld").warning("Unable to parse the " + PROPERTY_TRANSFER_PARALLELISM + " property as an integer: '" + parallelism + "', using " + transfer_parallelism + " instead");
+                }
+            }
         }
+        transferParallelism_ = transfer_parallelism;
     }
 
     /**
@@ -122,4 +147,15 @@ public class VersionResolution {
     public Map<String, Version> versionOverrides() {
         return versionOverrides_;
     }
+
+    /**
+     * Returns the number of artifact transfers that are performed in parallel,
+     * {@code 1} means transfers are sequential.
+     *
+     * @return the number of parallel artifact transfers
+     * @since 2.3.1
+     */
+    public int transferParallelism() {
+        return transferParallelism_;
+    }
 }

src/test/java/rife/bld/dependencies/TestDependencySet.java

@@ -4,10 +4,19 @@
  */
 package rife.bld.dependencies;
 
+import com.sun.net.httpserver.HttpServer;
 import org.junit.jupiter.api.Test;
+import rife.ioc.HierarchicalProperties;
+import rife.tools.FileUtils;
 import rife.tools.StringUtils;
 
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.file.Files;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.*;
 import static rife.bld.dependencies.RepositoryTestHelper.getNextRepository;
@@ -373,4 +382,93 @@ public class TestDependencySet {
                └─ org.json:json:20250107
             """), dependencies.generateTransitiveDependencyTree(VersionResolution.dummy(), ArtifactRetriever.instance(), RepositoryTestHelper.getNextRepositories(), compile, runtime));
     }
+
+    @Test
+    void testTransferIntoDirectoryParallel() throws Exception {
+        var max_concurrent_transfers = new AtomicInteger();
+        var server = createTransferServer(max_concurrent_transfers);
+        server.start();
+        var tmp = Files.createTempDirectory("transfers").toFile();
+        try {
+            var dependencies = createTransferDependencies();
+            var repositories = List.of(new Repository("http://localhost:" + server.getAddress().getPort() + "/"));
+
+            var artifacts = dependencies.transferIntoDirectory(new VersionResolution(null), ArtifactRetriever.instance(), repositories, tmp, tmp);
+
+            assertTransferredArtifacts(dependencies, artifacts, tmp);
+            assertTrue(max_concurrent_transfers.get() > 1, "expected concurrent transfers, max was " + max_concurrent_transfers.get());
+        } finally {
+            server.stop(0);
+            FileUtils.deleteDirectory(tmp);
+        }
+    }
+
+    @Test
+    void testTransferIntoDirectorySequential() throws Exception {
+        var max_concurrent_transfers = new AtomicInteger();
+        var server = createTransferServer(max_concurrent_transfers);
+        server.start();
+        var tmp = Files.createTempDirectory("transfers").toFile();
+        try {
+            var properties = new HierarchicalProperties();
+            properties.put(VersionResolution.PROPERTY_TRANSFER_PARALLELISM, "1");
+            var resolution = new VersionResolution(properties);
+            assertEquals(1, resolution.transferParallelism());
+
+            var dependencies = createTransferDependencies();
+            var repositories = List.of(new Repository("http://localhost:" + server.getAddress().getPort() + "/"));
+
+            var artifacts = dependencies.transferIntoDirectory(resolution, ArtifactRetriever.instance(), repositories, tmp, tmp);
+
+            assertTransferredArtifacts(dependencies, artifacts, tmp);
+            assertEquals(1, max_concurrent_transfers.get(), "expected sequential transfers, max was " + max_concurrent_transfers.get());
+        } finally {
+            server.stop(0);
+            FileUtils.deleteDirectory(tmp);
+        }
+    }
+
+    private static DependencySet createTransferDependencies() {
+        var dependencies = new DependencySet();
+        for (var i = 1; i <= 6; i++) {
+            dependencies.include(new Dependency("com.example", "artifact" + i, new VersionNumber(1, 0, 0)));
+        }
+        return dependencies;
+    }
+
+    private static HttpServer createTransferServer(AtomicInteger maxConcurrentTransfers)
+    throws IOException {
+        var active_transfers = new AtomicInteger();
+        var server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+        server.createContext("/", exchange -> {
+            var active = active_transfers.incrementAndGet();
+            maxConcurrentTransfers.accumulateAndGet(active, Math::max);
+            try {
+                // delay the response so that parallel transfers overlap
+                Thread.sleep(200);
+
+                var body = exchange.getRequestURI().getPath().getBytes();
+                exchange.sendResponseHeaders(200, body.length);
+                exchange.getResponseBody().write(body);
+                exchange.close();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            } finally {
+                active_transfers.decrementAndGet();
+            }
+        });
+        server.setExecutor(Executors.newCachedThreadPool());
+        return server;
+    }
+
+    private static void assertTransferredArtifacts(DependencySet dependencies, List<RepositoryArtifact> artifacts, File directory) {
+        assertEquals(dependencies.size(), artifacts.size());
+        var index = 0;
+        for (var dependency : dependencies) {
+            var filename = dependency.artifactId() + "-" + dependency.version() + ".jar";
+            assertTrue(artifacts.get(index).location().endsWith(filename), "expected artifact " + filename + " at index " + index);
+            assertTrue(new File(directory, filename).exists(), "expected file " + filename + " to be transferred");
+            ++index;
+        }
+    }
 }