Rxjava/RxAndroidでチェックチェーンを実装する方法

Question

私はRxJava/RxAndroidの新バージョンです。 RxJava/RxAndroidを使用して、次のケースを実装したい：まず、ネットワークからデータを取得し、データのチェックを行い、チェックが失敗した場合はメインスレッドにエラーを表示する。

you can see flow chart here!

私はいくつかのRxJava操作を試してみたが、そうする良い方法を見つけることができません。 誰かがこれについて私を助けることができますか？どうもありがとう！

私はこのケースについていくつかのテストコードを書いています（文字列をデータとして使用します）。これ以上の簡単な方法はありますか？あなたは、単にデータをチェックし、エラーを投げるためmap()を使用することができます

Observable.just(s) 
      .flatMap(new Function<String, ObservableSource<String>>() { 
     @Override 
     public ObservableSource<String> apply(final String s) throws Exception { 
      return Observable.create(new ObservableOnSubscribe<String>() { 
       @Override 
       public void subscribe(ObservableEmitter<String> e) throws Exception { 
        if(s.length() < 3){ 
         e.onError(new Throwable("len")); 
        }else{ 
         e.onNext(s); 
         e.onComplete(); 
        } 
       } 
      }); 
     } 
    }).flatMap(new Function<String, ObservableSource<String>>() { 
     @Override 
     public ObservableSource<String> apply(final String s) throws Exception { 
      return Observable.create(new ObservableOnSubscribe<String>() { 
       @Override 
       public void subscribe(ObservableEmitter<String> e) throws Exception { 
        if(s.startsWith("a")){ 
         e.onError(new Throwable("start")); 
        }else{ 
         e.onNext(s); 
         e.onComplete(); 
        } 
       } 
      }); 
     } 
    }).subscribeOn(AndroidSchedulers.mainThread()) 
      .doOnError(new Consumer<Throwable>() { 
     @Override 
     public void accept(Throwable throwable) throws Exception { 
      System.out.println("get error: " + throwable.getMessage()); 
     } 
    }).subscribe(new Consumer<String>() { 
     @Override 
     public void accept(String s) throws Exception { 
      System.out.println(s); 
     } 
    }); 

Answer 1

あなたはそれがここで必要とされていないflatMap()でそれを行うことができますが、：

Observable.just(s) 
      .map(new Function<String, String>() { 
        @Override 
        public String apply(@NonNull String s) throws Exception { 
         if (s.length() < 3) { 
          throw new Exception("len"); 
         } else if (s.startsWith("a")) { 
          throw new Exception("start"); 
         } 
         return s; 
        } 
       } 
      ) 
      .subscribe(new Consumer<String>() { 
          @Override 
          public void accept(@NonNull String s) throws Exception { 
           System.out.println(s); 
          } 
         }, new Consumer<Throwable>() { 
          @Override 
          public void accept(@NonNull Throwable throwable) throws Exception { 
           System.out.println("get error: " + throwable.getMessage(); 
          } 
         }); 

ここにあなたが放出された値をチェックし、単に投げますあなたの小切手に応じて適切な例外。

---

とにかく、あなたの例では、あなたがObservable.error()とObservable.just()を使用することができ、スルー合格/エラーを放出するために、お客様ご自身によるObservableを作成する必要はありません。

.flatMap(new Function<String, ObservableSource<?>>() { 
       @Override 
       public ObservableSource<?> apply(@NonNull String s) throws Exception { 
        if (s.length() < 3) { 
         return Observable.error(new Exception("len")); 
        } else if (s.startsWith("a")) { 
         return Observable.error(new Exception("start")); 
        } else { 
         return Observable.just(s); 
        } 
       } 
      }) 

また、your'eあなたのサブスクライバでonError()を処理しないでください（ただしdoOnError()にある）ので、OnErrorNotImplementedExceptionでクラッシュします。

Answer 2

あなたは少し（あなたがとにかく使用すべきではありません）Observable.create（）を排除することで、あなたのコードを簡素化することができます。

Observable.just(s) 
     .flatMap(new Function<String, ObservableSource<String>>() { 
      @Override 
      public ObservableSource<String> apply(final String s) throws Exception { 
       return s.length() < 3 ? Observable.error(new Throwable("len")) 
        : Observable.just(s); 
      } 
     }).flatMap(new Function<String, ObservableSource<String>>() { 
      @Override 
      public ObservableSource<String> apply(final String s) throws Exception { 
       return s.startsWith("a") ? Observable.error(new Throwable("start")) 
        : Observable.just(s); 
      } 
     }) 
     .subscribe(...) 

それともdoOnEachとグァバ前提条件を使用することができます。

Observable.just(s) 
    .doOnEach(s -> { 
     Preconditions.checkArgument(s >= 3, "len"); 
     Preconditions.checkArgument(!s.startsWith("a"), "start"); 
    }) 
    .subscribe(...)