43

I'm converting small project written in RxJava 1.x to Reactor 3.x. All is good, except that I could not find out how to replace flatMap(Observable::from) with an appropriate counterpart. I have Mono<List<String>> and I need to convert it to Flux<String>.

1

4 Answers 4

87

In Reactor 3, the from operator has been specialized into a few variants, depending on the original source (array, iterable, etc...).

Use yourMono.flatMapMany(Flux::fromIterable) in your case.

5
  • 2
    The return value of method Mono.flatMap is Mono, not Flux.
    – RJ.Hwang
    Mar 20, 2018 at 7:24
  • @SimonBaslé why does the member reference operator does not work in kotlin? ` Mono.just(listOfElements).flatMapMany(Flux::fromIterable)` //this does not work ` I have to write code like the one below Mono.just(listOfElements).flatMapMany{ Flux.fromIterable(it) } `
    – rhozet
    Nov 21, 2019 at 10:32
  • @rhozet no idea, that works in Java AFAIK so... question for kotlin compiler specialists? Nov 22, 2019 at 12:13
  • yes it works in Java, probably question for compiler specialists :)
    – rhozet
    Dec 1, 2019 at 11:08
  • I believe youtrack.jetbrains.com/issue/KT-13003 is why the function reference doesn't work in Kotlin
    – lxdr
    Apr 10, 2020 at 20:40
9

I think that probably Flux::mergeSequential static factory fits better here:

 Iterable<Mono<String>> monos = ...
 Flux<String> f = Flux.mergeSequential(monos);

This kind of merge (sequential) will maintain the ordering inside given source iterable, and will also subscribe/request eagerly from all participating sources (so more parallelization expected while computing mono results).

1
  • Note that Flux.mergeSequential subscribes to each source sequentially without waiting for completion in between. If you strictly need the next Mono to happen after the previous one has finished (e.g. you're publishing to an event queue and order matters), you should use Flux.concat(monos).
    – Pin
    May 16, 2020 at 18:48
4

Thanks Simon, I implemented something like this:

List<Object> dbObjects = ListObjectsBD();
    List<Dao> daos = mapperObjToDao(dbObjects);
    Flux<Dao> daoFlux = Mono.just(daos).flatMapMany(Flux::fromIterable);
1

Another way would be using flatMapIterable()

Transform the item emitted by this Mono into Iterable, then forward its elements into the returned Flux.

So it looks like this:

List<String> strings = new ArrayList<>();
strings.add("hello");
strings.add("world");
strings.add("!");

Mono<List<String>> monoList = Mono.just(strings);

Flux<String> flux = monoList.flatMapIterable(list -> list);

Or replace lambda list -> list with Function.identity()

Flux<String> flux = monoList.flatMapIterable(Function.identity());

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.