RxJava: Exécuter les secondes observables seulement si le premier jette une erreur et répéter de la première

Question

J'utilise retorift pour frapper getAricle api et obtenir la liste des articles liés à l'utilisateur. getArticle api lancera erreur si jeton passé est expiré si oui je dois appeler refreshToken api pour obtenir un nouveau jeton puis de nouveau je dois appeler le 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) 
      }) 

Modifier

Même si les réponses données a montré une certaine direction, mais ceux-ci ne sont pas une réponse directe à ma question. Voici comment résoudre mais je pense que ce peut être refactorisé dans beaucoup meilleur code

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

EDIT

Je refactorisé mon code à un peu plus de version plus propre

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

EDIT

S'il vous plaît vérifier ma réponse pour le code refactorisé final

Answer 1

J'ai implémenté exactement cela. Voici une version légèrement modifiée de ce code:

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

Vous pouvez utiliser cette méthode comme ceci:

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

Answer 2

Quel genre d'erreur que vous recevrez ?. Il semble que vous pourriez utiliser l'opérateur onErrorResumeNext.

Cet opérateur une fois que recevoir un throwable, vous permettent de retourner un Observable plutôt le throwable dans le 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); 

    } 

Vous pouvez voir d'autres exemples ici https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/errors/ObservableExceptions.java

Answer 3

Je vais vous donner une autre option à l'aide groupBy opérateur

/** 
* 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

Je résolu mon problème après avoir lu plus sur RxJava et voici comment je mis en œuvre il. Tout d'abord va retrofit jeter 4xx erreur à onError ou onNext\onSuccess dépend de la façon dont nous le définissons. Ex:

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

ce jetteront toutes les erreurs 4xx à onError et au lieu de Single<Article> si vous le définir comme Single<Response<Article>> alors toute la réponse du serveur, y compris 4xx ira à 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) 
       }) 

J'utilise defer comme une enveloppe autour de mes réels Observable parce que je veux recréer l'article aller chercher observable sur réessayer après jeton actualiser parce que je veux Preferences.getAuthToken() pour être appelé à nouveau car mon code de jeton d'actualisation stocke le jeton récupéré de préférence.

retry renvoie true si le HttpException est de 401 et pas essayé une nouvelle tentative plus de 2 fois