Handling Mono Inside Flux Flatmap

user_6714426 0
Viewed 6540

I have a Flux of strings. For each string, I have to make a remote call. But the problem is that the method that makes the remote call actually returns a Mono of the response (obviously since corresponding to a single request, there'll be a single response).

What should be the correct pattern to handle such cases? One solution I can think of is to make serial (or parallel) calls for the stream elements and reduce the responses to a single one and return.

Here's the code:

fluxObj.flatmap(a -> makeRemoteCall(a)//converts the Mono of the response to a Flux).reduce(...)

I am being unable to wrap my head around inside the flatmap.The makeRemoteCall method returns a Mono. But the flatmap returns a Flux of the response. First, why is this happening? Second, does it mean that the returned Flux contains a single response object (that was returned in the Mono)?

project-reactor functional-programming java
2 Answers

If the mapper Function returns a Mono, then it means that there will be (at most) one derived value for each source element in the Flux.

Having the Function return:

  • an empty Mono (eg. Mono.empty()) for a given value means that this source value is "ignored"
  • a valued Mono (like in your example) means that this source value is asynchronously mapped to another specific value
  • a Flux with several derived values for a given value means that this source value is asynchronously mapped to several values

For instance, given the following flatMap:

Flux.just("A", "B")
    .flatMap(v -> Mono.just("value" + v))

Subscribing to the above Flux<String> and printing the emitted elements would yield:

valueA
valueB

Another fun example: with delays, one can get out of order results. Like this:

Flux.just(300, 100)
    .flatMap(delay -> Mono.delay(Duration.ofMillis(delay))
                          .thenReturn(delay + "ms")
    )

would result in a Flux<String> that yields:

100ms
300ms
user_11134860

If you see documentation flatMap, you can found answers to your questions:

Transform the elements emitted by this Flux asynchronously into Publishers, then flatten these inner publishers into a single Flux, sequentially and preserving order using concatenation.

Long story in short, 

@Test
public void testFlux() {
    Flux<String> oneString = Flux.just("1");

    oneString
        .flatMap(s -> testMono(s))
        .collectList()
        .subscribe(integers -> System.out.println("elements:" + integers));       

}

private Mono<Integer> testMono(String s) {
    return Mono.just(Integer.valueOf(s + "0"));
}

mapper - s -> testMono(s) where testMono(s) is Publisher (in your case makeRemoteCall(a)), it transforms a type of my oneString to Integer.

I collected Flux result to List, and printed it. Console output:

elements:[10]

It means result Flux after flatMap operator contains just one element.

user_71856100
Related