java - rxJava 2 无法交付异常与线程池和 flatMap

Question

将 subscribeOn 与 flatMap 结合使用时，我得到了一个 UndeliverableException。这个最小的例子说明了这个问题:

@Test
public void flatMapException() throws ExecutionException, InterruptedException {
    SettableFuture<Boolean> f = SettableFuture.create();

    Observable.just(1,2).flatMap(x -> {
        return Observable.just(1).flatMap(z -> {
            if (z == 1) return Observable.error(new IOException("haha"));
            return Observable.just(1);
        }).subscribeOn(Schedulers.computation());
    }).onErrorReturnItem(1).subscribeOn(Schedulers.computation()).subscribe(
        x -> {}, 
        e -> {
            f.set(true);
        }, () -> {
            f.set(true);
        });

    assertEquals(true, f.get());
}

我预计在触发第一个错误后 Observable 会停止运行并取消订阅。

我只想在第一个 Observable 成功完成后订阅 flatMap 返回的第二个 Observable。

我如何用 RxJava 表达我的意图？

Answer 1

这似乎可行:

Observable.just(1, 2, 3)
  .map(x -> Observable.error(new Exception("e" + x)))
  .compose(xs -> Observable.concat(xs));