java - 用于组合映射的 RxJava 组合器？

Question

我有一个Map像这样的任务:

Map<Identifier, Observable<Progress>> tasks;

我想将其转换为:

Observable<Map<Identifier, Progress>>

其中每个 Map<Identifier, Progress> element包含每个任务的最新元素。

我想要新的Observable仅当所有任务完成时才完成，当任意一个任务失败时失败。

有一个组合器吗？

Answer 1

要改造单例Map<Identifier, Observable<Progress>> task至<Identifier, Progress>使用:

Observable.just(Identifier)
    .flatMap(Observable<Progress> progressObservable, 
        (Identifier, Progress, Pair<Identifier, Progress>) -> {create pair of elements})

要将这些任务组合在一起，请使用 combineLatest运算符

Observable.combineLatest(List<Observable<Pair<Identifier, Progress>>>, 
    (List<Pair<Identifier, Progress>>) -> {merge into `Map`})

我希望我能很好地理解你的想法。当然，用伪代码编写。

<小时/>

实现:

public static <T, S> Observable<Map<T, S>> zipMaps(final Map<T, Observable<S>> tasks) {
    Objects.requireNonNull(tasks, "tasks is null");
    return Observable.combineLatest(
        tasks.entrySet()
            .stream()
            .map(entry -> Observable.combineLatest(
                Observable.just(entry.getKey()),
                entry.getValue(),
                Pair::with))
            .collect(ImmutableList.toImmutableList()),
        xs -> Arrays.stream(xs)
            .map(x -> (Pair<T, S>)x)
            .collect(ImmutableMap.toImmutableMap(Pair::getValue0, Pair::getValue1)));
}