RxJava：只有在第一个抛出错误时重复执行第二个可观察事件，并且从第一个重复执行

Question

我正在使用retorift命中getAricle API并获取与用户相关的文章列表。 getArticle API将引发错误，如果令牌传递已过期，如果是的话我得叫refreshToken API来获取新的令牌，有一次我有打电话给getArticle API

ApiController.createRx().getArticle(token) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe({ response -> toast(response.body().url) }, { e -> 
       println(e.printStackTrace()) 
       if(e is HttpException && e.code() in arrayOf(401,403)){      
        //Here I want to call refresh tolken api 
        toast("Auth error") 
       } 
       else 
        toast(R.string.something_went_wrong) 
      }) 

编辑

即使给出答案显示了一些方向，但这些并不是我的问题的直接答案。这是怎么解决的，但我觉得这是可以重构为更好的代码

ApiController.createRx().getArticle(Preference.getToken()) 
      .flatMap { value -> 
       if (value.code() in arrayOf(403, 401)) { 
        ApiController.refreshToken() 
        ApiController.createRx().getArticle(Preference.getToken()) 
       } else Observable.just(value) 
      } 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe({ response -> println("Success") }, { e -> 
       e.printStackTrace() 
       toast(R.string.something_went_wrong) 
      }) 



fun refreshToken() { 
     val token:String?=ApiController.createRx().refreshToken(Preferences.getRefreshToken()).blockingFirst()?.body()?.token 
     if (token != null) Preferences.setAuthToken(token) 
    } 

编辑

我重构我的代码稍微清洁的版本

Observable.defer { ApiController.createRx().getArticle(Preferences.getToken()) } 
      .flatMap { 
       if (it.code() in arrayOf(401, 403)) { 
        ApiController.refreshToken() 
        Observable.error(Throwable()) 
       } else Observable.just(it) 
      } 
      .retry(1) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe({println("Success") }, { 
       it.printStackTrace() 
       toast(R.string.something_went_wrong) 
      }) 



fun refreshToken() { 
     var token: String? = null 
     try { 
      token = createRx().refreshToken(Preferences.getRefreshToken()).blockingFirst().body()!!.token 
     } catch (e: Exception) { 
      throw e 
     } 
     println("saving token") 
     if (token != null) Preferences.setAuthToken(token) 
    } 

编辑

请检查我的答案为最终重构代码

Answer 1

我已经实现了这个确切的东西。下面是该代码稍加修改的版本：

private Observable<Object> refreshTokenIfNotAuthorized(Observable<? extends Throwable> errors) { 
    final AtomicBoolean alreadyRetried = new AtomicBoolean(false); 

    return errors.flatMap(error -> { 

     boolean isAuthorizationError = /* some logic analyzing each error*/ ; 

     if (isAuthorizationError && !alreadyRetried.get()) { 
      try { 
       alreadyRetried.set(true); 
       String newToken = federatedTokenRefresher.refreshToken() 
                 .toBlocking() 
                 .first(); 

       setLogin(newToken); 
       return Observable.just(null); 

      } catch (Exception e) { 
       return Observable.error(error); 
      } 

     } 
     return Observable.error(error); 
    }); 
} 

您可以使用此方法，像这样：

doSomethingRequiringAuth().retryWhen(this::refreshTokenIfNotAuthorized); 

Answer 2

你会收到什么样的错误？这似乎可以使用onErrorResumeNext运营商。

该运营商一旦收到一个抛出，让您在onerror的返回可观察到的，而不是抛出

@Test 
    public void observableOnErrorResumeException() { 
     Integer[] numbers = {0, 1, 2, 3, 4, 5}; 

     Observable.from(numbers) 
       .doOnNext(number -> { 
        if (number > 3) { 
         try { 
          throw new IllegalArgumentException(); 
         } catch (Exception e) { 
          throw new RuntimeException(e); 
         } 
        } 

       }) 
       .onErrorResumeNext(t -> Observable.just(666)) 
       .subscribe(System.out::println); 

    } 

你可以看到更多在这里https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/errors/ObservableExceptions.java

Answer 3

的例子，我会给你使用GROUPBY另一种选择运营商

/** 
* In this example we create a response code group. 
*/ 
@Test 
public void testGroupByCode() { 
    Observable.from(Arrays.asList(401,403, 200)) 
      .groupBy(code -> code) 
      .subscribe(groupByCode -> { 
       switch (groupByCode.getKey()) { 
        case 401: { 
         System.out.println("refresh token"); 
         processResponse(groupByCode); 
         break; 
        } 
        case 403: { 
         System.out.println("refresh token"); 
         processResponse(groupByCode); 
         break; 
        } 
        default: { 
         System.out.println("Do the toast"); 
         processResponse(groupByCode); 
        } 
       } 
      }); 
} 

private void processResponse(GroupedObservable<Integer, Integer> groupByCode) { 
    groupByCode.asObservable().subscribe(value -> System.out.println("Response code:" + value)); 
} 

Answer 4

我解决我的问题阅读更多关于RxJava后，这是我如何实现它。 首先将retrofit丢到4xx错误onError或onNext\onSuccess取决于我们如何定义它。 例如：

@GET("content") fun getArticle(@Header("Authorization") token: String):Single<Article>

这将引发所有的4xx错误onError和替代Single<Article>如果你将它定义为Single<Response<Article>>那么所有从服务器包括4XX的响应会去onNext\onSuccess

Single.defer { ApiController.createRx().getArticle(Preferences.getAuthToken())} 
       .doOnError { 
        if (it is HttpException && it.code() == 401) 
         ApiController.refreshToken() 
       } 
       .retry { attempts, error -> attempts < 3 && error is HttpException && error.code() == 401 } 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe({println("Success") }, { 
        it.printStackTrace() 
        toast(R.string.something_went_wrong) 
       }) 

我正在使用defer作为我的实际Observable的包装，因为我想在重新标记刷新后重试文章fetch observable，因为我想要Preferences.getAuthToken()将再次被调用，因为我的刷新令牌代码优先存储新提取的令牌。

retry返回true如果HttpException是401，而不是试图重试2次以上