Reactor 将 Flux 应用于其他 Flux 的每个发射？

【问题标题】：Reactor apply Flux to each emission of other Flux?Reactor 将 Flux 应用于其他 Flux 的每个发射？
【发布时间】：2019-03-13 18:11:53
【问题描述】：

我有两个 Flux 对象，例如：

Flux<Item> 和 Flux<Transformation>

data class Item(val value: Int)

data class Transformation(val type: String, val value: Int)

我想对每个项目应用所有转换 - 例如：

var item = Item(15)

val transformations = listOf(Transformation(type = "MULTIPLY", value = 8), ...)

transformations.forEach {
  if (it.type == "MULTIPLY") {
    item = Item(item.value * it.value) 
  }
}

但是当有Flux'es 的Item 和Transformation 时

【问题讨论】：

标签： kotlin project-reactor

【解决方案1】：

您可以使用java.util.function.UnaryOperator 代替Transformation 类。希望这个 Java 示例可以帮助您：

@Test
public void test() {
    Flux<Item> items = Flux.just(new Item(10), new Item(20));
    Flux<UnaryOperator<Item>> transformations = Flux.just(
            item -> new Item(item.value * 8),
            item -> new Item(item.value - 3));

    Flux<Item> transformed = items.flatMap(item -> transformations
            .collectList()
            .map(unaryOperators -> transformFunction(unaryOperators)
                    .apply(item)));

    System.out.println(transformed.collectList().block());
}

Function<Item, Item> transformFunction(List<UnaryOperator<Item>> itemUnaryOperators) {
    Function<Item, Item> transformFunction = UnaryOperator.identity();
    for (UnaryOperator<Item> itemUnaryOperator : itemUnaryOperators) {
        transformFunction = transformFunction.andThen(itemUnaryOperator);
    }
    return transformFunction;
}

【讨论】：

唯一的问题是项目被重复的转换次数所复制，这不是我想要的
你想为每个项目链接转换吗？
确切地说，我想将所有转换应用于每个项目（例如，一项和十次转换应该产生一项，但应用了十次转换）
那么转换序列呢？有关系吗？
我已经编辑了答案。但恐怕转换的顺序会在实际系统中发挥重要作用。所以也许，最好有一个转换列表