Reactor using map() on a Flux does not push elements but .flapMap() does

user_6596263 0
Viewed 656

I suppose even after reading the javadocs multiple times I don't get the difference between map and flatMap apart from the synchronous vs asynchronous transforms.In the following code I don't get any events (it behaves as if the subscribe() was not there.

final Flux<GroupedFlux<String, TData>> groupedFlux =
            flux.groupBy(Event::getPartitionKey);
    groupedFlux.subscribe(g -> g.delayElements(Duration.ofMillis(100))
            .map(this::doWork)
            .doOnError(throwable -> log.error("error: ", throwable))
            .onErrorResume(e -> Mono.empty())
            .subscribe());

However a flapMap() works. This works fine -

final Flux<GroupedFlux<String, TData>> groupedFlux =
            flux.groupBy(Event::getPartitionKey);
    groupedFlux.subscribe(g -> g.delayElements(Duration.ofMillis(100))
            .flatMap(this::doWork)
            .doOnError(throwable -> log.error("error: ", throwable))
            .onErrorResume(e -> Mono.empty())
            .subscribe());

Why is that?

EDIT: Added sample code for the doWork method as suggested in a comment.

private Mono<Object> doWork(Object event) {
    // do some work and possibly return Mono<Object> or
    return Mono.empty();
}
project-reactor
1 Answers

I think it's because your doWork method returns a Mono. The map operation implicitly wraps your returned object inside a Mono, so you get a Mono<Mono>. Since your original flow subscribes to the wrapper Mono, but the one inside that one is not subscribed to it never produces anything. In contrast flatMap needs the wrapping to be explicit.

Try modifing your doWork method to return not a Mono and do the explicit Mono.just in the flatMap operation.

user_111035500
Related