feat(webflux): Ensure batch processing execution sequence.

Ahoo-Wang · Nov 17, 2024 · 016ee96 · 016ee96
1 parent 1beb04a
commit 016ee96
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/wow-webflux/src/main/kotlin/me/ahoo/wow/webflux/route/event/state/ResendStateEventHandler.kt b/wow-webflux/src/main/kotlin/me/ahoo/wow/webflux/route/event/state/ResendStateEventHandler.kt
@@ -43,7 +43,7 @@ class ResendStateEventHandler(
     fun handle(afterId: String, limit: Int): Mono<BatchResult> {
         val target = CompensationTarget(function = RESEND_FUNCTION)
         return snapshotRepository.scanAggregateId(aggregateMetadata.namedAggregate, afterId, limit)
-            .flatMap { aggregateId ->
+            .concatMap { aggregateId ->
                 stateEventCompensator.resend(
                     aggregateId = aggregateId,
                     target = target,

diff --git a/.../main/kotlin/me/ahoo/wow/webflux/route/snapshot/BatchRegenerateSnapshotHandlerFunction.kt b/.../main/kotlin/me/ahoo/wow/webflux/route/snapshot/BatchRegenerateSnapshotHandlerFunction.kt
@@ -49,7 +49,7 @@ class BatchRegenerateSnapshotHandlerFunction(
             namedAggregate = aggregateMetadata.namedAggregate,
             afterId = afterId,
             limit = limit,
-        ).flatMap { aggregateId ->
+        ).concatMap { aggregateId ->
             handler.handle(aggregateId).thenReturn(aggregateId)
         }.toBatchResult(afterId).toServerResponse(request, exceptionHandler)
     }