Deep "flatReduce" for WebFlux?

user_1111753 0
Viewed 174

I have defined an operator flatReduce() which is to reduce() what flatMap() is to map():

public class FlatReduce {
    public static <V, W> Mono<V> flatReduce(Mono<V> initial, Iterable<W> items, BiFunction<V, W, Mono<V>> accumulator) {
        for (W item : items) {
            initial = initial.flatMap(v -> accumulator.apply(v, item));
        }
        return initial;
    }

    public static void main(String[] args) {
        flatReduce(Mono.just(1), IntStream.range(0, 4000).mapToObj(it -> it).collect(toList()), (a, b) -> Mono.just(a + b)).block();
    }
}

This yields deeply nested flatMaps and I have observed stack overflows. Is there any way to work around this (maybe turning this into continuation style)?

Thanks.

reactive-streams project-reactor spring-webflux spring-boot
1 Answers

Found a workaround by converting the Mono to CompletableFuture.

    @Test
    public void runTest() {
        System.out.println(flatReduce(Mono.just(1), IntStream.range(1, 500000).mapToObj(it -> it).collect(toList()), (a, b) -> Mono.just(a + 1)).block());
    }

    private static <V, W> Mono<V> flatReduce(Mono<V> initial, List<W> items, BiFunction<V, W, Mono<V>> accumulator) {
        return Mono.fromCompletionStage(flatReduceWithFuture(initial.toFuture(), items, (v, w) -> accumulator.apply(v, w).toFuture()));
    }

    private static <V, W> CompletableFuture<V> flatReduceWithFuture(CompletableFuture<V> initial, List<W> items, BiFunction<V, W, CompletableFuture<V>> accumulator) {
        for (W item : items) {
            initial = initial.thenCompose(x -> accumulator.apply(x, item));
        }
        return initial;
    }
user_11117530
Related