fix(modrinth): correctly retrieve specific dependency version info

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

@@ -54,6 +54,10 @@ public <T> ObjectFetchBuilder<T> toObject(Class<T> type) {
         return new ObjectFetchBuilder<>(this.config, type);
     }
 
+    public <T> ObjectListFetchBuilder<T> toObjectList(Class<T> type) {
+        return new ObjectListFetchBuilder<>(this.config, type);
+    }
+
     public <T> ObjectFetchBuilder<T> toObject(Class<T> type, ObjectMapper objectMapper) {
         return new ObjectFetchBuilder<>(this.config, type, objectMapper);
     }

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

@@ -0,0 +1,38 @@
+package me.itzg.helpers.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import java.util.List;
+import me.itzg.helpers.json.ObjectMappers;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.core5.http.HttpHeaders;
+import org.apache.hc.core5.http.message.BasicHttpRequest;
+
+public class ObjectListFetchBuilder<T> extends FetchBuilder<ObjectListFetchBuilder<T>> {
+
+    private final Class<T> type;
+    private final ObjectMapper objectMapper;
+
+    ObjectListFetchBuilder(Config config, Class<T> type, ObjectMapper objectMapper) {
+        super(config);
+        this.type = type;
+        this.objectMapper = objectMapper;
+    }
+
+    ObjectListFetchBuilder(Config config, Class<T> type) {
+        this(config, type, ObjectMappers.defaultMapper());
+    }
+
+    @Override
+    protected void configureRequest(BasicHttpRequest request) throws IOException {
+        super.configureRequest(request);
+        request.addHeader(HttpHeaders.ACCEPT, "application/json");
+    }
+
+    public List<T> execute() throws IOException {
+        try (CloseableHttpClient client = client()) {
+            return client.execute(get(), new ObjectListMapperHandler<>(type, objectMapper));
+        }
+    }
+
+}

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

@@ -0,0 +1,24 @@
+package me.itzg.helpers.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import java.io.IOException;
+import java.util.List;
+import org.apache.hc.core5.http.HttpEntity;
+
+public class ObjectListMapperHandler<T> extends LoggingResponseHandler<List<T>> {
+
+    private final ObjectMapper objectMapper;
+    private final Class<T> type;
+
+    public ObjectListMapperHandler(Class<T> type, ObjectMapper objectMapper) {
+        this.type = type;
+        this.objectMapper = objectMapper;
+    }
+
+    @Override
+    public List<T> handleEntity(HttpEntity entity) throws IOException {
+        final ObjectReader objectReader = objectMapper.readerForListOf(type);
+        return objectReader.readValue(entity.getContent());
+    }
+}

src/main/java/me/itzg/helpers/modrinth/ModrinthCommand.java

@@ -1,8 +1,11 @@
 package me.itzg.helpers.modrinth;
 
+import static me.itzg.helpers.http.Fetch.fetch;
+
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.IOException;
+import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Instant;
@@ -12,10 +15,9 @@
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import lombok.extern.slf4j.Slf4j;
 import me.itzg.helpers.Manifests;
-import me.itzg.helpers.http.HttpClientException;
-import me.itzg.helpers.http.ReactorNettyBits;
 import me.itzg.helpers.http.Uris;
 import me.itzg.helpers.json.ObjectMappers;
 import me.itzg.helpers.modrinth.model.DependencyType;
@@ -27,9 +29,6 @@
 import picocli.CommandLine.Command;
 import picocli.CommandLine.ExitCode;
 import picocli.CommandLine.Option;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 
 @Command(name = "modrinth", description = "Automates downloading of modrinth resources")
 @Slf4j
@@ -39,8 +38,6 @@ public class ModrinthCommand implements Callable<Integer> {
     };
     public static final String MODS_SUBDIR = "mods";
 
-    private final ReactorNettyBits bits = new ReactorNettyBits();
-
     private final String baseUrl;
 
     @Option(names = "--projects", description = "Project ID or Slug", required = true, split = ",", paramLabel = "id|slug")
@@ -61,7 +58,6 @@ public class ModrinthCommand implements Callable<Integer> {
     @Option(names = "--allowed-version-type", defaultValue = "release", description = "Valid values: ${COMPLETION-CANDIDATES}")
     VersionType versionType;
 
-    final Set<Path> outputFiles = Collections.synchronizedSet(new HashSet<>());
     final Set<String/*projectId*/> projectsProcessed = Collections.synchronizedSet(new HashSet<>());
 
     @SuppressWarnings("unused")
@@ -89,25 +85,9 @@ public Integer call() throws Exception {
             oldManifest = null;
         }
 
-        Flux.fromIterable(projects)
-            .parallel()
-            .runOn(Schedulers.parallel())
-            .doOnNext(projectRef -> log.debug("Starting with projectRef={}", projectRef))
-            .flatMap(this::getProject)
-            .doOnNext(project -> projectsProcessed.add(project.getId()))
-            .flatMap(project ->
-                getVersionsForProject(project.getId())
-                    .mapNotNull(this::pickVersion)
-                    .doOnNext(version -> log.debug("Picked version={} for project={}", version, project))
-                    .expand(this::expandDependencies)
-                    .mapNotNull(this::pickVersionFile)
-                    .doOnNext(versionFile -> log.debug("VersionFile={}", versionFile))
-                    .flatMap(versionFile -> download(project.getProjectType(), versionFile))
-                    .doOnNext(outputFiles::add)
-                    .doOnNext(path -> log.debug("Wrote file={} for project={}", path, project))
-            )
-            .then()
-            .block();
+        final List<Path> outputFiles = projects.stream()
+            .flatMap(this::processProject)
+            .collect(Collectors.toList());
 
         final Manifest newManifest = Manifest.builder()
             .timestamp(Instant.now())
@@ -131,32 +111,51 @@ public Integer call() throws Exception {
         return ExitCode.OK;
     }
 
-    private Flux<Version> expandDependencies(Version version) {
-        return Flux.fromStream(version.getDependencies().stream()
-                .filter(versionDependency ->
-                    projectsProcessed.add(versionDependency.getProjectId()) &&
-                        (versionDependency.getDependencyType() == DependencyType.required ||
-                            downloadOptionalDependencies && versionDependency.getDependencyType() == DependencyType.optional)
-                )
+    private Stream<Version> expandDependencies(Version version) {
+        log.debug("Expanding dependencies of version={}", version);
+        return version.getDependencies().stream()
+            .filter(dep ->
+                projectsProcessed.add(dep.getProjectId()) &&
+                    (dep.getDependencyType() == DependencyType.required ||
+                        downloadOptionalDependencies && dep.getDependencyType() == DependencyType.optional)
             )
-            .flatMap(versionDependency -> {
-                if (versionDependency.getVersionId() == null) {
-                    return getVersionsForProject(versionDependency.getProjectId())
-                        .mapNotNull(this::pickVersion);
-                } else {
-                    return getVersion(versionDependency.getVersionId());
+            .flatMap(dep -> {
+                try {
+                    final Version depVersion;
+                    if (dep.getVersionId() == null) {
+                        log.debug("Fetching versions of dep={} and picking", dep);
+                        depVersion = pickVersion(
+                            getVersionsForProject(dep.getProjectId())
+                        );
+                    } else {
+                        log.debug("Fetching version for dep={}", dep);
+                        depVersion = getVersion(dep.getVersionId());
+                    }
+                    if (depVersion != null) {
+                        log.debug("Resolved version={} for dep={}", depVersion, dep);
+                        return Stream.concat(
+                                Stream.of(depVersion),
+                                expandDependencies(depVersion)
+                            )
+                            .peek(expandedVer -> {
+                                log.debug("Expanded dependency={} into version={}", dep, expandedVer);
+                            });
+                    } else {
+                        return Stream.empty();
+                    }
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
                 }
-            })
-            .doOnNext(depVersion -> log.debug("Expanded depVersion={} from version={}", depVersion, version));
+            });
+
     }
 
-    private Mono<Version> getVersion(String versionId) {
-        return bits.jsonClient()
-            .get()
-            .uri(Uris.populate(
-                baseUrl + "/versions/{id}", versionId
-            ))
-            .responseSingle(bits.readInto(Version.class));
+    private Version getVersion(String versionId) throws IOException {
+        return fetch(Uris.populateToUri(
+            baseUrl + "/version/{id}", versionId
+        ))
+            .toObject(Version.class)
+            .execute();
     }
 
     private Version pickVersion(List<Version> versions) {
@@ -168,7 +167,7 @@ private Version pickVersion(List<Version> versions) {
         return null;
     }
 
-    private Mono<Path> download(ProjectType projectType, VersionFile versionFile) {
+    private Path download(ProjectType projectType, VersionFile versionFile) {
         if (log.isDebugEnabled()) {
             log.debug("Downloading {}", versionFile);
         } else {
@@ -182,41 +181,14 @@ private Mono<Path> download(ProjectType projectType, VersionFile versionFile) {
             .resolve(MODS_SUBDIR)
             .resolve(versionFile.getFilename());
 
-        if (Files.exists(outPath)) {
-            log.debug("Output file={} already exists", outPath);
-            return Mono.just(outPath);
+        try {
+            return fetch(URI.create(versionFile.getUrl()))
+                .toFile(outPath)
+                .skipExisting(true)
+                .execute();
+        } catch (IOException e) {
+            throw new RuntimeException(e);
         }
-
-        return bits.client()
-            .followRedirect(true)
-            .get()
-            .uri(versionFile.getUrl())
-            .responseContent()
-            .aggregate()
-            .asInputStream()
-            .publishOn(Schedulers.boundedElastic())
-            .map(inputStream -> {
-                try {
-                    try {
-                        Files.createDirectories(outPath.getParent());
-                    } catch (IOException e) {
-                        throw new RuntimeException("Unable to create directory for downloaded file", e);
-                    }
-
-                    Files.copy(inputStream, outPath);
-
-                    return outPath;
-                } catch (IOException e) {
-                    throw new RuntimeException(e);
-                } finally {
-                    try {
-                        inputStream.close();
-                    } catch (IOException e) {
-                        log.warn("Unable to close download aggregate stream", e);
-                    }
-                }
-            })
-            .doOnNext(outputFiles::add);
     }
 
     private VersionFile pickVersionFile(Version version) {
@@ -230,28 +202,54 @@ private VersionFile pickVersionFile(Version version) {
         }
     }
 
-    private Mono<Project> getProject(String projectIdOrSlug) throws HttpClientException {
-        return bits.jsonClient()
-            .get()
-            .uri(Uris.populate(
+    private Project getProject(String projectIdOrSlug) {
+        try {
+            return fetch(Uris.populateToUri(
                 baseUrl + "/project/{id|slug}",
                 projectIdOrSlug
             ))
-            .responseSingle(bits.readInto(Project.class));
+                .toObject(Project.class)
+                .execute();
+        } catch (IOException e) {
+            throw new RuntimeException("Getting project " + projectIdOrSlug, e);
+        }
     }
 
-    private Mono<List<Version>> getVersionsForProject(String project) {
-        return bits.jsonClient()
-            .get()
-            .uri(Uris.populate(
+    private List<Version> getVersionsForProject(String project) {
+        try {
+            return fetch(Uris.populateToUri(
                 baseUrl + "/project/{id|slug}/version?loaders={loader}&game_versions={gameVersion}",
                 project, arrayOfQuoted(loader.toString()), arrayOfQuoted(gameVersion)
             ))
-            .responseSingle(bits.readInto(VERSION_LIST));
+                .toObjectList(Version.class)
+                .execute();
+        } catch (IOException e) {
+            throw new RuntimeException("Getting versions for project " + project, e);
+        }
     }
 
 
     private String arrayOfQuoted(String value) {
         return "[\"" + value + "\"]";
     }
+
+    private Stream<? extends Path> processProject(String projectRef) {
+        log.debug("Starting with projectRef={}", projectRef);
+
+        final Project project = getProject(projectRef);
+        if (projectsProcessed.add(project.getId())) {
+            final List<Version> versions = getVersionsForProject(project.getId());
+            final Version version = pickVersion(versions);
+
+            if (version != null) {
+                return Stream.concat(
+                        Stream.of(version),
+                        expandDependencies(version)
+                    )
+                    .map(this::pickVersionFile)
+                    .map(versionFile -> download(project.getProjectType(), versionFile));
+            }
+        }
+        return Stream.empty();
+    }
 }