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()); 
} 

私は最初のエラーの後に観察可能で実行が停止し、登録解除を解雇されていることを期待しました。

最初の正常終了後にflatMapで返された2番目のObservableを購読したいと思います。

RxJavaで私の意図を表現するにはどうすればよいですか？

Answer 1

これは動作するようです：

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