reactor - 从列表中收集成功和失败的单声道

Question

我有一个 Mono 列表，其中一些完成了一些失败：

Mono<String> success1 = Mono.just("hello");
Mono<String> success2 = Mono.just("world");
Mono<String> error1 = Mono.error(new IllegalArgumentException("error1"));
Mono<String> error2 = Mono.error(new IllegalArgumentException("error2"));
List<Mono<String>> monos = Arrays.asList(success1, success2, error1, error2);

从这个列表中，如何一次性收集成功和失败的Monos？（意味着每个 Mono 只发射一次）

private Mono<Result> collectMonoListResult(List<Mono<String>> monos) {
   // TODO what comes here
   // so that the result will have "hello" and "world" in the successList
   // and 2 IllegalArgumentException instances on the errorList?
}

@AllArgsConstructor
class Result {
   private List<String> successList;
   private List<Throwable> errorList;
}

Answer 1

一种解决方案是具体化每个 Mono，将它们合并为 Flux，并收集结果。

    private <T> Mono<Result<T>> collectMonoListResult(List<Mono<T>> monos) {
        return monos.stream()
                // Materialize each Mono<T> into a Mono<Signal<T>>
                // to be able to obtain the emitted element or error.
                .map(Mono::materialize)
                // Merge all the Mono<Signal<T>> into a Flux<Signal<T>>
                .<Flux<Signal<T>>>reduce(Flux.empty(), Flux::mergeWith, Flux::mergeWith)
                // Collect all the Signals into a single Result
                .collect(Result::new, (result, signal) -> {
                    if (signal.isOnNext()) {
                        result.addSuccess(signal.get());
                    } else if (signal.isOnError()) {
                        result.addError(signal.getThrowable());
                    }
                });
    }

    public static class Result<T> {
        private final List<T> successList = new ArrayList<>();
        private final List<Throwable> errorList = new ArrayList<>();

        public void addSuccess(T success) {
            successList.add(success);
        }
        public void addError(Throwable error) {
            errorList.add(error);
        }

        public List<T> getSuccessList() {
            return Collections.unmodifiableList(successList);
        }

        public List<Throwable> getErrorList() {
            return Collections.unmodifiableList(errorList);
        }
    }

reactor - 从列表中收集成功和失败的单声道

reactor - collect both successful and failed monos from list

java

project-reactor