RxJavaでエラーが発生した後にアイテムをストリーミングする方法はありますか？

Question

私はRxJava初心者です。次の問題があります。 私はアイテムのシーケンスを持っていると、アイテムの上にエラーを伝播すると、それを無視して他のアイテムの処理を続行したいとします。私はスニペット以下のいる

：

Observable.from(Arrays.asList("1", "2", "3")) 
      .map(x -> { 
       if (x.equals("2")) { 
        throw new NullPointerException(); 
       } 
       return x + "-"; 
      }) 
      .onExceptionResumeNext(Observable.empty()) 
      .subscribe(System.out::println); 

私は取得しています：1 -

しかし、私は取得したい：1、3-

私はそれをどのように行うことができますか？

Answer 1

このトリックは、次の例のように、何らかの形で変換された値を新しい観測可能なフラットマップにラップすることです。 flatMapの各値は、例外をスローして、値で値を処理できるようになりました。 flatMapのサブストリームを1つの要素のみで構成すると、onErrorの後に観測対象が閉じられるかどうかは関係ありません。私はテスト環境としてRxJava2を使用します。

@Test 
public void name() throws Exception { 
    Observable<String> stringObservable = Observable.fromArray("1", "2", "3") 
      .flatMap(x -> { 
       return Observable.defer(() -> { 
        try { 
         if (x.equals("2")) { 
          throw new NullPointerException(); 
         } 
         return Observable.just(x + "-"); 
        } catch (Exception ex) { 
         return Observable.error(ex); 
        } 
       }).map(s -> { 
        if (s.equals("3-")) { 
         throw new IllegalArgumentException(); 
        } 
        return s + s; 
       }).take(1) 
         .zipWith(Observable.just("X"), (s, s2) -> s + s2) 
         .onErrorResumeNext(Observable.empty()); 
      }); 

    TestObserver<String> test = stringObservable.test(); 

    test.assertResult("1-1-X"); 
} 

の