任意数量的Observable:：flatMap调用链

Question

比如说，我有几个可观察的供应商对get通话有副作用：

Subject<String> firstSubject = PublishSubject.create();
Supplier<Observable<String>> firstSupplier = () -> {
    System.out.println("side effect of first one");
    return firstSubject;
};

Subject<String> secondSubject = PublishSubject.create();
Supplier<Observable<String>> secondSupplier = () -> {
    System.out.println("side effect of second one");
    return secondSubject;
};

Subject<String> thirdSubject = PublishSubject.create();
Supplier<Observable<String>> thirdSupplier = () -> {
    System.out.println("side effect of third one");
    return thirdSubject;
};

现在我想用以下方式将它们结合起来——在前一个供应商的可观察的发出一个值后，调用下一个供应商的get（onNext）。我可以通过以下代码来实现：

firstSupplier.get()
        .flatMap(__ -> secondSupplier.get())
        .flatMap(__ -> thirdSupplier.get())
        .subscribe();
// output: side effect of first one
firstSubject.onNext("");
// output: side effect of second one
secondSubject.onNext("");
// output: side effect of third one

我如何重写这段代码来接受未知数量的供应商，作为-例如-集合传递

我已经回顾了可观察的的各种工厂方法（比如合并，concat），但是他们都在收集可观察的资源，这意味着我必须给我所有的供应商打电话获取。然而，在我的例子中，重要的是要懒洋洋地调用它——只有在前面的Observable发出一个值之后。




Answer 1

编辑2：

我完全忘记了Observable.flatMap调用映射器，即使maxConcurrency设置为1并将生成的可观察排队等待以后运行。

我希望下面的设置能像预期的那样工作（即，它没有指定对象的后续onNext应该发生什么）。

Subject<String> firstSubject = PublishSubject.create();
Supplier<Observable<String>> firstSupplier = () -> {
    System.out.println("side effect of first one");
    return firstSubject;
};

Subject<String> secondSubject = PublishSubject.create();
Supplier<Observable<String>> secondSupplier = () -> {
    System.out.println("side effect of second one");
    return secondSubject;
};

Subject<String> thirdSubject = PublishSubject.create();
Supplier<Observable<String>> thirdSupplier = () -> {
    System.out.println("side effect of third one");
    return thirdSubject;
};

Collection<Supplier<Observable<String>>> collection =
    Arrays.asList(firstSupplier, secondSupplier, thirdSupplier);

Observable.fromIterable(collection)
.concatMap(supplier -> supplier.get().take(1))
.subscribe();

System.out.println("// output: side effect of first one");
firstSubject.onNext("");
System.out.println("// output: side effect of second one");
secondSubject.onNext("");
System.out.println("// output: side effect of third one");

版画：

side effect of first one
// output: side effect of first one
side effect of second one
// output: side effect of second one
side effect of third one
// output: side effect of third one