将 Flux 排放与重复合并？

Question

我有一个Flux发出项目的：

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

考虑 2 个“篮子”和 4 个“鸡蛋”排放。我想将这些排放物合并为两个：每个包含一个“篮子”和 4 个“鸡蛋”：

有人知道这种转变吗？Flux 是有限的，不应超过 1K 项。

编辑：

到目前为止我取得的成就 - 我将排放分为GroupedFlux. 现在我需要将GroupedFluxcontains Basket1,Basket2与第二个包含 'Eggs'结合起来，以便在每个篮子中生成两个带有“重复”鸡蛋的篮子。

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

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

所需流量： Flux.just("Basket1(Egg1,Egg2, Egg3, Egg4)","Basket2(Egg1,Egg2, Egg3, Egg4)")

Answer 1

flatMap您可以使用和来实现此结果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);
  }
}