perf: replace concatMap to flatMapSequential to improve parallelism a…

…nd efficiency
halo-dev · Sep 26, 2024 · 54ae723 · 54ae723
1 parent f6409a0
commit 54ae723
Show file tree

Hide file tree

Showing 20 changed files with 28 additions and 28 deletions.
diff --git a/application/src/main/java/run/halo/app/content/comment/CommentServiceImpl.java b/application/src/main/java/run/halo/app/content/comment/CommentServiceImpl.java
@@ -58,7 +58,7 @@ public Mono<ListResult<ListedComment>> listComment(CommentQuery commentQuery) {
                 commentQuery.toPageRequest())
             .flatMap(comments -> Flux.fromStream(comments.get()
                     .map(this::toListedComment))
-                .concatMap(Function.identity())
+                .flatMapSequential(Function.identity())
                 .collectList()
                 .map(list -> new ListResult<>(comments.getPage(), comments.getSize(),
                     comments.getTotal(), list)

diff --git a/application/src/main/java/run/halo/app/content/comment/ReplyServiceImpl.java b/application/src/main/java/run/halo/app/content/comment/ReplyServiceImpl.java
@@ -166,7 +166,7 @@ public Mono<ListResult<ListedReply>> list(ReplyQuery query) {
         return client.listBy(Reply.class, query.toListOptions(), query.toPageRequest())
             .flatMap(list -> Flux.fromStream(list.get()
                     .map(this::toListedReply))
-                .concatMap(Function.identity())
+                .flatMapSequential(Function.identity())
                 .collectList()
                 .map(listedReplies -> new ListResult<>(list.getPage(), list.getSize(),
                     list.getTotal(), listedReplies))

diff --git a/application/src/main/java/run/halo/app/content/impl/PostServiceImpl.java b/application/src/main/java/run/halo/app/content/impl/PostServiceImpl.java
@@ -77,7 +77,7 @@ public Mono<ListResult<ListedPost>> listPost(PostQuery query) {
             )
             .flatMap(listResult -> Flux.fromStream(listResult.get())
                 .map(this::getListedPost)
-                .concatMap(Function.identity())
+                .flatMapSequential(Function.identity())
                 .collectList()
                 .map(listedPosts -> new ListResult<>(listResult.getPage(), listResult.getSize(),
                     listResult.getTotal(), listedPosts)
@@ -175,7 +175,7 @@ private Flux<Contributor> listContributors(List<String> usernames) {
             return Flux.empty();
         }
         return Flux.fromIterable(usernames)
-            .concatMap(userService::getUserOrGhost)
+            .flatMapSequential(userService::getUserOrGhost)
             .map(user -> {
                 Contributor contributor = new Contributor();
                 contributor.setName(user.getMetadata().getName());

diff --git a/application/src/main/java/run/halo/app/content/impl/SinglePageServiceImpl.java b/application/src/main/java/run/halo/app/content/impl/SinglePageServiceImpl.java
@@ -88,7 +88,7 @@ public Flux<ListedSnapshotDto> listSnapshots(String pageName) {
     public Mono<ListResult<ListedSinglePage>> list(SinglePageQuery query) {
         return client.listBy(SinglePage.class, query.toListOptions(), query.toPageRequest())
             .flatMap(listResult -> Flux.fromStream(listResult.get().map(this::getListedSinglePage))
-                .concatMap(Function.identity())
+                .flatMapSequential(Function.identity())
                 .collectList()
                 .map(listedSinglePages -> new ListResult<>(
                     listResult.getPage(),

diff --git a/application/src/main/java/run/halo/app/core/user/service/impl/DefaultAttachmentService.java b/application/src/main/java/run/halo/app/core/user/service/impl/DefaultAttachmentService.java
@@ -71,7 +71,7 @@ public Mono<Attachment> upload(
                     .map(configMap -> new UploadOption(filePart, policy, configMap));
             })
             .flatMap(uploadContext -> extensionGetter.getExtensions(AttachmentHandler.class)
-                .concatMap(handler -> handler.upload(uploadContext))
+                .flatMapSequential(handler -> handler.upload(uploadContext))
                 .next())
             .switchIfEmpty(Mono.error(() -> new ServerErrorException(
                 "No suitable handler found for uploading the attachment.", null)))
@@ -113,7 +113,7 @@ public Mono<Attachment> delete(Attachment attachment) {
             .flatMap(policy -> client.get(ConfigMap.class, policy.getSpec().getConfigMapName())
                 .map(configMap -> new DeleteOption(attachment, policy, configMap)))
             .flatMap(deleteOption -> extensionGetter.getExtensions(AttachmentHandler.class)
-                .concatMap(handler -> handler.delete(deleteOption))
+                .flatMapSequential(handler -> handler.delete(deleteOption))
                 .next());
     }
 
@@ -122,7 +122,7 @@ public Mono<URI> getPermalink(Attachment attachment) {
         return client.get(Policy.class, attachment.getSpec().getPolicyName())
             .flatMap(policy -> client.get(ConfigMap.class, policy.getSpec().getConfigMapName())
                 .flatMap(configMap -> extensionGetter.getExtensions(AttachmentHandler.class)
-                    .concatMap(handler -> handler.getPermalink(attachment, policy, configMap))
+                    .flatMapSequential(handler -> handler.getPermalink(attachment, policy, configMap))
                     .next()
                 )
             );
@@ -133,7 +133,7 @@ public Mono<URI> getSharedURL(Attachment attachment, Duration ttl) {
         return client.get(Policy.class, attachment.getSpec().getPolicyName())
             .flatMap(policy -> client.get(ConfigMap.class, policy.getSpec().getConfigMapName())
                 .flatMap(configMap -> extensionGetter.getExtensions(AttachmentHandler.class)
-                    .concatMap(handler -> handler.getSharedURL(attachment, policy, configMap, ttl))
+                    .flatMapSequential(handler -> handler.getSharedURL(attachment, policy, configMap, ttl))
                     .next()
                 )
             );

diff --git a/...ication/src/main/java/run/halo/app/extension/router/ExtensionCompositeRouterFunction.java b/...ication/src/main/java/run/halo/app/extension/router/ExtensionCompositeRouterFunction.java
@@ -47,7 +47,7 @@ public ExtensionCompositeRouterFunction(ReactiveExtensionClient client,
     @NonNull
     public Mono<HandlerFunction<ServerResponse>> route(@NonNull ServerRequest request) {
         return Flux.fromIterable(getRouterFunctions())
-            .concatMap(routerFunction -> routerFunction.route(request))
+            .flatMapSequential(routerFunction -> routerFunction.route(request))
             .next();
     }
 

diff --git a/application/src/main/java/run/halo/app/plugin/DefaultPluginRouterFunctionRegistry.java b/application/src/main/java/run/halo/app/plugin/DefaultPluginRouterFunctionRegistry.java
@@ -34,7 +34,7 @@ public DefaultPluginRouterFunctionRegistry() {
     public Mono<HandlerFunction<ServerResponse>> route(@NonNull ServerRequest request) {
         var secureRequest = new SecureServerRequest(request);
         return Flux.fromIterable(this.routerFunctions)
-            .concatMap(routerFunction -> routerFunction.route(secureRequest))
+            .flatMapSequential(routerFunction -> routerFunction.route(secureRequest))
             .next();
     }
 

diff --git a/application/src/main/java/run/halo/app/plugin/PluginServiceImpl.java b/application/src/main/java/run/halo/app/plugin/PluginServiceImpl.java
@@ -247,7 +247,7 @@ public Flux<DataBuffer> uglifyJsBundle() {
             });
         var body = Flux.fromIterable(startedPlugins)
             .sort(Comparator.comparing(PluginWrapper::getPluginId))
-            .concatMap(pluginWrapper -> {
+            .flatMapSequential(pluginWrapper -> {
                 var pluginId = pluginWrapper.getPluginId();
                 return Mono.<Resource>fromSupplier(
                         () -> BundleResourceUtils.getJsBundleResource(
@@ -274,7 +274,7 @@ public Flux<DataBuffer> uglifyJsBundle() {
     public Flux<DataBuffer> uglifyCssBundle() {
         return Flux.fromIterable(pluginManager.getStartedPlugins())
             .sort(Comparator.comparing(PluginWrapper::getPluginId))
-            .concatMap(pluginWrapper -> {
+            .flatMapSequential(pluginWrapper -> {
                 var pluginId = pluginWrapper.getPluginId();
                 var dataBufferFactory = DefaultDataBufferFactory.sharedInstance;
                 return Mono.<Resource>fromSupplier(() -> BundleResourceUtils.getJsBundleResource(

diff --git a/application/src/main/java/run/halo/app/plugin/extensionpoint/DefaultExtensionGetter.java b/application/src/main/java/run/halo/app/plugin/extensionpoint/DefaultExtensionGetter.java
@@ -86,10 +86,10 @@ private <T extends ExtensionPoint> Flux<T> getEnabledExtensions(String epdName,
                 }
                 var extensions = getExtensions(extensionPoint).cache();
                 return Flux.fromIterable(extensionDefNames)
-                    .concatMap(extensionDefName ->
+                    .flatMapSequential(extensionDefName ->
                         client.fetch(ExtensionDefinition.class, extensionDefName)
                     )
-                    .concatMap(extensionDef -> {
+                    .flatMapSequential(extensionDef -> {
                         var className = extensionDef.getSpec().getClassName();
                         return extensions.filter(
                             extension -> Objects.equals(extension.getClass().getName(),

diff --git a/application/src/main/java/run/halo/app/theme/dialect/GlobalHeadInjectionProcessor.java b/application/src/main/java/run/halo/app/theme/dialect/GlobalHeadInjectionProcessor.java
@@ -71,7 +71,7 @@ protected void doProcess(ITemplateContext context, IModel model,
 
         // apply processors to modelToInsert
         getTemplateHeadProcessors(context)
-            .concatMap(processor -> processor.process(
+            .flatMapSequential(processor -> processor.process(
                 new SecureTemplateContext(context), modelToInsert, structureHandler)
             )
             .then()

diff --git a/application/src/main/java/run/halo/app/theme/dialect/TemplateFooterElementTagProcessor.java b/application/src/main/java/run/halo/app/theme/dialect/TemplateFooterElementTagProcessor.java
@@ -58,7 +58,7 @@ protected void doProcess(ITemplateContext context, IProcessableElementTag tag,
         modelToInsert.add(context.getModelFactory().createText(globalFooterText));
 
         getTemplateFooterProcessors(context)
-            .concatMap(processor -> processor.process(
+            .flatMapSequential(processor -> processor.process(
                 new SecureTemplateContext(context), tag, structureHandler, modelToInsert)
             )
             .then()

diff --git a/application/src/main/java/run/halo/app/theme/finders/impl/CommentPublicQueryServiceImpl.java b/application/src/main/java/run/halo/app/theme/finders/impl/CommentPublicQueryServiceImpl.java
@@ -86,7 +86,7 @@ public Mono<ListResult<CommentVo>> list(Ref ref, PageRequest pageParam) {
                 return client.listBy(Comment.class, listOptions, pageRequest)
                     .flatMap(listResult -> Flux.fromStream(listResult.get())
                         .map(this::toCommentVo)
-                        .concatMap(Function.identity())
+                        .flatMapSequential(Function.identity())
                         .collectList()
                         .map(commentVos -> new ListResult<>(listResult.getPage(),
                             listResult.getSize(),
@@ -102,7 +102,7 @@ public Mono<ListResult<CommentVo>> list(Ref ref, PageRequest pageParam) {
     public Mono<ListResult<CommentWithReplyVo>> convertToWithReplyVo(ListResult<CommentVo> comments,
         int replySize) {
         return Flux.fromIterable(comments.getItems())
-            .concatMap(commentVo -> {
+            .flatMapSequential(commentVo -> {
                 var commentName = commentVo.getMetadata().getName();
                 return listReply(commentName, 1, replySize)
                     .map(replyList -> CommentWithReplyVo.from(commentVo)
@@ -135,7 +135,7 @@ public Mono<ListResult<ReplyVo>> listReply(String commentName, PageRequest pageP
                     .orElse(PageRequestImpl.ofSize(0));
                 return client.listBy(Reply.class, listOptions, pageRequest)
                     .flatMap(list -> Flux.fromStream(list.get().map(this::toReplyVo))
-                        .concatMap(Function.identity())
+                        .flatMapSequential(Function.identity())
                         .collectList()
                         .map(replyVos -> new ListResult<>(list.getPage(), list.getSize(),
                             list.getTotal(),

diff --git a/application/src/main/java/run/halo/app/theme/finders/impl/ContributorFinderImpl.java b/application/src/main/java/run/halo/app/theme/finders/impl/ContributorFinderImpl.java
@@ -33,6 +33,6 @@ public Flux<ContributorVo> getContributors(List<String> names) {
             return Flux.empty();
         }
         return Flux.fromIterable(names)
-            .concatMap(this::getContributor);
+            .flatMapSequential(this::getContributor);
     }
 }
diff --git a/application/src/main/java/run/halo/app/theme/finders/impl/PostFinderImpl.java b/application/src/main/java/run/halo/app/theme/finders/impl/PostFinderImpl.java
@@ -291,7 +291,7 @@ public Mono<ListResult<PostArchiveVo>> archives(Integer page, Integer size, Stri
     public Flux<ListedPostVo> listAll() {
         return postPredicateResolver.getListOptions()
             .flatMapMany(listOptions -> client.listAll(Post.class, listOptions, defaultSort()))
-            .concatMap(postPublicQueryService::convertToListedVo);
+            .flatMapSequential(postPublicQueryService::convertToListedVo);
     }
 
     static int pageNullSafe(Integer page) {

diff --git a/application/src/main/java/run/halo/app/theme/finders/impl/PostPublicQueryServiceImpl.java b/application/src/main/java/run/halo/app/theme/finders/impl/PostPublicQueryServiceImpl.java
@@ -67,7 +67,7 @@ public Mono<ListResult<ListedPostVo>> list(ListOptions queryOptions, PageRequest
             })
             .flatMap(listOptions -> client.listBy(Post.class, listOptions, page))
             .flatMap(list -> Flux.fromStream(list.get())
-                .concatMap(post -> convertToListedVo(post)
+                .flatMapSequential(post -> convertToListedVo(post)
                     .flatMap(postVo -> populateStats(postVo)
                         .doOnNext(postVo::setStats).thenReturn(postVo)
                     )

diff --git a/...cation/src/main/java/run/halo/app/theme/finders/impl/SinglePageConversionServiceImpl.java b/...cation/src/main/java/run/halo/app/theme/finders/impl/SinglePageConversionServiceImpl.java
@@ -140,7 +140,7 @@ public Mono<ListResult<ListedSinglePageVo>> listBy(ListOptions listOptions,
 
         return client.listBy(SinglePage.class, rewroteListOptions, rewrotePageRequest)
             .flatMap(list -> Flux.fromStream(list.get())
-                .concatMap(this::convertToListedVo)
+                .flatMapSequential(this::convertToListedVo)
                 .collectList()
                 .map(pageVos ->
                     new ListResult<>(list.getPage(), list.getSize(), list.getTotal(), pageVos)

diff --git a/application/src/main/java/run/halo/app/theme/finders/impl/TagFinderImpl.java b/application/src/main/java/run/halo/app/theme/finders/impl/TagFinderImpl.java
@@ -48,7 +48,7 @@ public Mono<TagVo> getByName(String name) {
     @Override
     public Flux<TagVo> getByNames(List<String> names) {
         return Flux.fromIterable(names)
-            .concatMap(this::getByName);
+            .flatMapSequential(this::getByName);
     }
 
     @Override

diff --git a/application/src/main/java/run/halo/app/theme/router/SinglePageRoute.java b/application/src/main/java/run/halo/app/theme/router/SinglePageRoute.java
@@ -65,7 +65,7 @@ public class SinglePageRoute
     @NonNull
     public Mono<HandlerFunction<ServerResponse>> route(@NonNull ServerRequest request) {
         return Flux.fromIterable(routerFunctions())
-            .concatMap(routerFunction -> routerFunction.route(request))
+            .flatMapSequential(routerFunction -> routerFunction.route(request))
             .next();
     }
 

diff --git a/application/src/main/java/run/halo/app/theme/router/ThemeCompositeRouterFunction.java b/application/src/main/java/run/halo/app/theme/router/ThemeCompositeRouterFunction.java
@@ -55,7 +55,7 @@ public class ThemeCompositeRouterFunction implements RouterFunction<ServerRespon
     @NonNull
     public Mono<HandlerFunction<ServerResponse>> route(@NonNull ServerRequest request) {
         return Flux.fromIterable(cachedRouters)
-            .concatMap(routerFunction -> routerFunction.route(request))
+            .flatMapSequential(routerFunction -> routerFunction.route(request))
             .next();
     }
 

diff --git a/application/src/main/java/run/halo/app/theme/router/factories/PostRouteFactory.java b/application/src/main/java/run/halo/app/theme/router/factories/PostRouteFactory.java
@@ -131,7 +131,7 @@ private Mono<ServerResponse> postResponse(ServerRequest request,
     Mono<String> determineTemplate(ServerRequest request, PostVo postVo) {
         return Flux.fromIterable(defaultIfNull(postVo.getCategories(), List.of()))
             .filter(category -> isNotBlank(category.getSpec().getPostTemplate()))
-            .concatMap(category -> viewNameResolver.resolveViewNameOrDefault(request,
+            .flatMapSequential(category -> viewNameResolver.resolveViewNameOrDefault(request,
                 category.getSpec().getPostTemplate(), null)
             )
             .next()