fetch: download specific file to adjacent temp

Author: itzg

src/main/java/me/itzg/helpers/files/ReactiveFileUtils.java

@@ -5,8 +5,10 @@
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.time.Instant;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import reactor.core.publisher.Mono;
@@ -107,4 +109,13 @@ public static <T> Mono<T> removeFailedDownload(Throwable throwable, Path outputF
             })
             .subscribeOn(Schedulers.boundedElastic());
     }
+
+    public static Function<Path, Mono<Path>> moveTo(Path to) {
+        return from -> Mono.fromCallable(() -> {
+                log.debug("Moving {} to {}", from, to);
+                Files.move(from, to, StandardCopyOption.REPLACE_EXISTING);
+                return to;
+            })
+            .subscribeOn(Schedulers.boundedElastic());
+    }
 }

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

@@ -64,6 +64,7 @@ public Mono<Path> assemble() {
 
         final boolean useIfModifiedSince = skipUpToDate && Files.exists(file);
 
+        final Path tempDownloadFile = file.getParent().resolve(file.getFileName() + ".download");
         return useReactiveClient(client ->
             client
                 .doOnRequest((httpClientRequest, connection) ->
@@ -107,7 +108,7 @@ public Mono<Path> assemble() {
                         return failedContentTypeMono(resp);
                     }
 
-                    return ReactiveFileUtils.writeByteBufFluxToFile(byteBufFlux, file)
+                    return ReactiveFileUtils.writeByteBufFluxToFile(byteBufFlux, tempDownloadFile)
                         .flatMap(fileSize -> {
                             statusHandler.call(FileDownloadStatus.DOWNLOADED, uri, file);
                             downloadedHandler.call(uri, file, fileSize);
@@ -120,12 +121,13 @@ public Mono<Path> assemble() {
                                             uri, formatDuration(durationMillis), transferRate(durationMillis, fileSize)
                                         );
                                     }
-                                    return Mono.just(file);
+                                    return Mono.just(tempDownloadFile);
                                 });
                         });
 
                 })
                 .last()
+                .flatMap(ReactiveFileUtils.moveTo(file))
                 .contextWrite(context -> context.put("downloadStart", currentTimeMillis()))
         );
     }