Chaining nested Async actions from Multi to Uni...

[rafaeltuelho]

I've an unusual scenario where I need to chain a set of actions (async) starting from a Multi<T> and ending by returning a Uni<Void>...

  @ReactiveTransactional
  @WithSpan("HeroService.replaceAllHeroes")
  public Uni<Void> replaceAllHeroes(@SpanAttribute("arg.heroes") List<Hero> heroes) {
    Log.debug("Replacing all heroes");
		Multi<Hero> multiHeroes = Multi.createFrom().items(heroes.stream());
                 // >>> HERE I HAD TO FORCE THIS CAST <<<
		return (Uni<Void>) multiHeroes.onItem().transformToUni(hero -> {
			return this.filteredSetOfPowers(hero).onItem().transformToUni(set -> {
				hero.updatePowers(set); // this is a entity sync method
				return deleteAllHeroes().replaceWith(this.heroRepository.persist(heroes)); // this returns Uni<Void>
			});
		});

  }

full code here: https://github.com/rafaeltuelho/quarkus-super-heroes/blob/fighting-with-super-power/rest-heroes/src/main/java/io/quarkus/sample/superheroes/hero/service/HeroService.java

I can't understand why Muti.transformToUni() return a MultiFlatten<O, T>...

I would appreciate it if someone could help me understand why I'm getting a MultiFlatten<O, T> here. i woudl expect Multi.tranformToUni infer from my last nested return type, which is a Uni<Void> in my case...

[jponge]

transformToUni needs a strategy to deal with items: do you want items to be emitted in the same order as the upstream or not?

See https://smallrye.io/smallrye-mutiny/2.0.0/tutorials/transforming-items-asynchronously/#transforming-items-from-multi-the-merge-vs-concatenate-dilemma and you can also use shortcut methods: transformToUniAndMerge and transformToUniAndConcatenate

[rafaeltuelho]

@jponge thanks for this reference.

I learned the Multi.toUni() method and tried to refactor differently:

  @ReactiveTransactional
  @WithSpan("HeroService.replaceAllHeroes")
  public Uni<Void> replaceAllHeroes(@SpanAttribute("arg.heroes") List<Hero> heroes) {
    Log.debug("Replacing all heroes");
    Multi<Hero> multiHeroes = Multi.createFrom().items(heroes.stream());
    
    return multiHeroes.onItem().transform(hero -> {
        return this.mergePowers(hero).onItem().invoke(mergedPowers -> {
          hero.updatePowers(mergedPowers);
        });
    }).toUni().replaceWith(deleteAllHeroes().replaceWith(this.heroRepository.persist(heroes)));