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>
.
-
See baeldung.com/java-mono-list-to-flux– Ahmad AbdelghanyOct 24, 2022 at 12:09
4 Answers
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.
-
2
-
@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) } `– rhozetNov 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
-
-
I believe youtrack.jetbrains.com/issue/KT-13003 is why the function reference doesn't work in Kotlin– lxdrApr 10, 2020 at 20:40
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).
-
Note that
Flux.mergeSequential
subscribes to each source sequentially without waiting for completion in between. If you strictly need the nextMono
to happen after the previous one has finished (e.g. you're publishing to an event queue and order matters), you should useFlux.concat(monos)
.– PinMay 16, 2020 at 18:48
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);
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());