如何實現Rxjava/RxAndroid的檢查鏈

Question

我是RxJava/RxAndroid的新手。我想使用RxJava/RxAndroid來實現以下情況：首先，從網絡獲取數據，然後對數據進行一些檢查，如果任何檢查失敗，只顯示Main Thread中的錯誤。

you can see flow chart here!

我嘗試一些RxJava操作，但無法找到一個很好的方式這樣做。 有人可以幫我嗎？非常感謝！

我寫了一些關於這種情況的測試代碼（使用String作爲數據），還有更簡單的方法嗎？

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()這裏是不是需要它這樣做，你可以簡單地使用map()檢查數據，並引發錯誤：

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用於發射的錯誤/直通通過創建，您可以使用Observable.error()和Observable.just()：

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

而且，傻冒不在您的訂戶處處理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(...)