RxJava + Retrofit + polling

Question

Ho una chiamata di aggiornamento e desidero richiamarla ogni 30 secondi. Per fare questo io uso un Observable.interval(0, 30, TimeUnit.SECONDS)

Observable 
    .interval(0, 30, TimeUnit.SECONDS) 
    .flatMap(x -> RestApi.instance().getUsers()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(list -> { 
        // ... 
       }, 
       error -> Timber.e(error, "can't load users")); 

Il mio problema: se la chiamata API non riesce, onError si chiama e le unsubscribes sottoscrizione e per il polling non funziona più :-(

Per catturare l'errore api ho aggiunto un retryWhen

Observable 
    .interval(0, 30, TimeUnit.SECONDS) 
    .flatMap(x -> RestApi.instance().getUsers() 
         .retryWhen(errors -> errors 
          .flatMap(error -> Observable.timer(15, TimeUnit.SECONDS)))) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(list -> { 
        // ... 
       }, 
       error -> Timber.e(error, "can't load users")); 

Questo cattura l'errore, ma ottengo più chiamate API nel tempo. Ogni 30 secondi ottengo un nuovo segnale sondaggio che termina in una nuova richiesta di api. ma se la richiesta API non riesce esso si ritenta . Così io avere una nuova richiesta più tutti i tentativi.

La mia domanda: come posso gestire un errore API senza annullare l'iscrizione dal segnale di polling?

Answer 1

Leggi come usare correttamente retryWhen e repeatWhen. http://blog.danlew.net/2016/01/25/rxjavas-repeatwhen-and-retrywhen-explained/

E come utilizzare gli operatori onError: http://blog.danlew.net/2015/12/08/error-handling-in-rxjava/

E 'davvero facile w Rx :) Non ho intenzione di darvi una soluzione definitiva, solo giocare con essa e cercare di capire il flusso qui.

Answer 2

È possibile utilizzare onErrorResumeNext o onExceptionResumeNext e passare il valore "errore". È possibile cercare altri movimentazioni di errore dipende dalle vostre esigenze here

Answer 3

Se si desidera la richiesta di non finire in onError quando una richiesta non riesce, invece di tornare Observable<yourUserType> getUsers(), ne fanno restituire un Observable<Response<yourUserType>> getUsers(). In questo modo sarai in grado di intercettare l'errore di rete nell'oggetto Response.

Questo metodo funziona solo se si utilizza retrofit 2.x

Answer 4

È possibile utilizzare questo codice, in questo codice attuare i numeri tentativi e intervallo di tempo tra richiesta

private static int COUNTER_START = 1; 
private static final int ATTEMPTS = 6; 
private static final int ORIGINAL_DELAY_IN_SECONDS = 2; 

remoteData.getAllRides(idSearch) 
      .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() { 
         @Override 
         public Observable<?> call(Observable<? extends Void> observable) { 
          return observable.flatMap(new Func1<Void, Observable<?>>() { 
           @Override 
           public Observable<?> call(Void aVoid) { 
            if(COUNTER_START > ATTEMPTS){ 
             throw new RuntimeException(); 
            } 
            COUNTER_START++; 
            return Observable.timer(ORIGINAL_DELAY_IN_SECONDS, TimeUnit.SECONDS); 
           } 
          }); 
         } 
        }) 
      .takeUntil(new Func1<RideResponse, Boolean>() { 
       @Override 
       public Boolean call(RideResponse rideResponse) { 
        return rideResponse.getState().equals("finished");//this is the validation finish polling 

       } 
      }).filter(new Func1<RideResponse, Boolean>() { 
       @Override 
       public Boolean call(RideResponse rideResponse) { 
        return rideResponse.getState().equals("finished"); //this is the validation finish polling 
       } 
      }).map(rideResponse -> Log.e("",rideResponse.toString())) 
     .doOnError(err -> Log.e("Polling", "Error retrieving messages: " + err));