Merge Flux emissions with duplication?

Question

I'm having a Flux that emits items:

data class Item(
  val isEgg: Boolean,
  val isBasket: Boolean
)

Consider 2 'baskets' and 4 'eggs' emissions. I would like to merge those emissions into two: each containing one 'basket' and 4 'eggs':

Is someone aware of such transformation? Flux is finite and should not exceed 1K items.

EDIT:

What I achieved so far - I grouped emissions into GroupedFlux. Now I would need to combine GroupedFlux containing Basket1, Basket2 with with second containing 'Eggs' in order to produce two baskets with "duplicated" eggs in each one.

    val flux = Flux.just("Egg1", "Egg2", "Basket1", "Egg3", "Egg4", "Basket2")

    val block = flux.groupBy {
        it.startsWith("Egg")
    }

Desired Flux: Flux.just("Basket1(Egg1,Egg2, Egg3, Egg4)","Basket2(Egg1,Egg2, Egg3, Egg4)")

Answer 1

You can achieve this result with flatMap and reduce:

void combine() {
  Flux<String> flux = 
    Flux.just("Egg1", "Egg2", "Basket1", "Egg3", "Egg4", "Basket2");
  Flux<String> eggs = flux.filter(str -> str.startsWith("Egg"));
  Flux<String> basketNames = flux.filter(str -> str.startsWith("Basket"));
  basketNames.flatMap(basketName -> eggs.reduce(
      new Basket(basketName),
      (basket, egg) -> {
        basket.add(egg);
        return basket;
      })
  );
}

class Basket {

  private final String name;
  private final List<String> eggs;

  Basket(final String name) {
    this.name = name;
    this.eggs = new ArrayList<>();
  }

  public void add(String egg) {
    eggs.add(egg);
  }
}