迭代一个Flux，里面执行一个Mono，下一步使用结果

Question

我想做与下面类似的事情，其中​​上一次调用的结果将用于使用 Project Reactor 对同一服务的下一次调用。

Message current;
Message next;
for each (Step step in steps)
{
    current = new Message(step, next);
    next = execute(current);
}

这就是我正在尝试使用反应器做的事情：

1. 对于每个“步骤”（不断变化）

   一个。为该步骤和最后一个结果创建一条消息（以 null 开始）。

   b.使用消息调用服务并获取结果（单声道）。

   c。将最后一条消息设置为此结果，以便在 1a 中使用。

2. 获取最后的结果

到目前为止，我在这方面的拙劣尝试看起来像：

return fromIterable(request.getPipeline())
    .map(s -> PipelineMessage.builder()
        .client(client)
        .step(s.getStep())
        .build())
    .flatMap(z -> {
        return this.pipelineService.execute(z);
    })
    .last()
    .map(m -> ok()
        .entity(m.getPayload())
        .type(m.getType())
        .build());

Answer 1

不确定您的确切要求。但我认为reduce 功能在这里可以提供帮助。

Flux<String> stringFlux = Flux.fromIterable(Arrays.asList("a", "b", "c"));

stringFlux
        .reduce("empty", (next, step) -> {
            String current = message(step, next);
            return execute(current);
        })
        .map(String::toUpperCase)
        .subscribe(System.out::println);

这里的message和execute是这样的函数。

String message(String step, String next){
    return "message[" + step + ":" + next + "]";
}

String execute(String current){
    return "execute(" + current + ")";
}

最终输出将是最后执行的消息。

EXECUTE(MESSAGE[C:EXECUTE(MESSAGE[B:EXECUTE(MESSAGE[A:EMPTY])])])

这里的初始步骤不能是null。相反，您可以使用一个空的步骤对象并将其视为null。

---

Flux<String> stringFlux = Flux.fromIterable(Arrays.asList("a", "b", "c"));

stringFlux
        .reduce(Flux.just("empty"), (Flux<String> next, String step) -> {
            Flux<String> current = message(step, next);
            return current.flatMap(this::execute);
        })
        .flatMapMany(a -> a)
        .subscribe(System.out::println);


Flux<String> message(String step, Flux<String> next){
    return next.map(v -> "message(" + v + ":" + step + ")");
}

Mono<String> execute(String current){
    return Mono.just("execute(" + current + ")");
}

输出：

execute(message(execute(message(execute(message(empty:a)):b)):c))