project-reactor - 将通量发射与重复合并？

Question

我有一个Flux发出项目的：

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

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

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

编辑：

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

    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)")

标签： project-reactor

Answer 1

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);
  }
}