refactor: ensure each subscription happens on a dedicated single-thre…

…aded worker for mono callable
halo-dev · May 26, 2023 · 4a97dd6 · 4a97dd6
1 parent a32d3d3
commit 4a97dd6
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 1 deletion.
diff --git a/application/src/main/java/run/halo/app/core/extension/endpoint/PluginEndpoint.java b/application/src/main/java/run/halo/app/core/extension/endpoint/PluginEndpoint.java
@@ -226,6 +226,7 @@ private Mono<ServerResponse> upgradeFromUri(ServerRequest request) {
             .flatMap(upgradeRequest -> Mono.fromCallable(() -> DataBufferUtils.toInputStream(
                 reactiveUrlDataBufferFetcher.fetch(upgradeRequest.uri())))
             )
+            .subscribeOn(Schedulers.boundedElastic())
             .onErrorMap(throwable -> {
                 log.error("Failed to fetch plugin file from uri.", throwable);
                 return new ThemeUpgradeException("Failed to fetch plugin file from uri.", null,
@@ -247,6 +248,7 @@ private Mono<ServerResponse> installFromUri(ServerRequest request) {
             .flatMap(installRequest -> Mono.fromCallable(() -> DataBufferUtils.toInputStream(
                 reactiveUrlDataBufferFetcher.fetch(installRequest.uri())))
             )
+            .subscribeOn(Schedulers.boundedElastic())
             .doOnError(throwable -> {
                 log.error("Failed to fetch plugin file from uri.", throwable);
                 throw new ThemeInstallationException("Failed to fetch plugin file from uri.", null,
@@ -605,6 +607,6 @@ private Mono<Path> transferToTemp(InputStream inputStream) {
             Path tempFile = Files.createTempFile("halo-plugins", ".jar");
             FileCopyUtils.copy(inputStream, Files.newOutputStream(tempFile));
             return tempFile;
-        });
+        }).subscribeOn(Schedulers.boundedElastic());
     }
 }
diff --git a/application/src/main/java/run/halo/app/core/extension/theme/ThemeEndpoint.java b/application/src/main/java/run/halo/app/core/extension/theme/ThemeEndpoint.java
@@ -36,6 +36,7 @@
 import org.springframework.web.server.ServerWebInputException;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.util.retry.Retry;
 import run.halo.app.core.extension.Setting;
 import run.halo.app.core.extension.Theme;
@@ -247,6 +248,7 @@ private Mono<ServerResponse> upgradeFromUri(ServerRequest request) {
             .flatMap(upgradeRequest -> Mono.fromCallable(() -> DataBufferUtils.toInputStream(
                 reactiveUrlDataBufferFetcher.fetch(upgradeRequest.uri())))
             )
+            .subscribeOn(Schedulers.boundedElastic())
             .doOnError(throwable -> {
                 log.error("Failed to fetch zip file from uri.", throwable);
                 throw new ThemeUpgradeException("Failed to fetch zip file from uri.", null,
@@ -268,6 +270,7 @@ private Mono<ServerResponse> installFromUri(ServerRequest request) {
             .flatMap(installRequest -> Mono.fromCallable(() -> DataBufferUtils.toInputStream(
                 reactiveUrlDataBufferFetcher.fetch(installRequest.uri())))
             )
+            .subscribeOn(Schedulers.boundedElastic())
             .doOnError(throwable -> {
                 log.error("Failed to fetch zip file from uri.", throwable);
                 throw new ThemeInstallationException("Failed to fetch zip file from uri.", null,