RxJava: вызывать сеть несколько раз с увеличением задержки, игнорируя ошибки

Question

Я хочу создать своего рода экспоненциальное отключение для сетевого вызова. Но вместо обычного интервала отсрочки мой сетевой вызов следует повторять несколько раз (с увеличением задержки), а затем быть полным - независимо от результата сетевого вызова.

Например, вариант по умолчанию должен выглядеть следующим образом:

→ wait 1s → network call → onNext(result) 
    → wait 3s → network call → onNext(result) 
    → wait 10s → network call → onNext(result) 
    → onCompleted 

Когда некоторые из сетевых вызовов имеют ошибки, эти ошибки должны быть зарегистрированы, но не прервать саму задачу. Скажем, второй сетевой вызов имеет ошибку, то она должна выглядеть следующим образом:

→ wait 1s → network call → onNext(result) 
    → wait 3s → network call error → log error 
    → wait 10s → network call → onNext(result) 
    → onCompleted 

И когда все вызовы имеют ошибки:

→ wait 1s → network call error → log error 
    → wait 3s → network call error → log error 
    → wait 10s → network call error → log error 
    → onCompleted 

У меня уже есть наблюдаемом, что можно назвать сеть, где код выглядит следующим образом:

public Observable<Stuff> loadStuff() { 
    Request request = new Request.Builder() 
     .url("http://example.com/stuff").build(); 
    return myCall(request, Stuff.class); 
} 

public <T> Observable<T> myCall(Request request, Class<T> resultClass) { 
    // calls OkHttp3 and parses the result 
    return okhttp(request) 
      .flatMap(parseResponse(resultClass)); 
} 

Как я неоднократно использовать Observable от loadStuff() для достижения моей желаемого интервала отсрочки?

Answer 1

Похоже, вы могли бы использовать RetryWhen, а также onErrorResumeNext, чтобы избежать разрыва трубопровода, в RetryWhen вы можете вернуться к наблюдаемому с экспоненциальной задержкой добиться того, что вы хотите.

 @Test 
public void observableOnErrorResumeNext() { 
    Subscription subscription = Observable.just(null) 
              .map(Object::toString) 
              .doOnError(failure -> System.out.println("Error:" + failure.getCause())) 
              .retryWhen(errors -> errors.doOnNext(o -> count++) 
                    .flatMap(t -> count > 3 ? Observable.error(t) : Observable.just(null)), 
                Schedulers.newThread()) 
              .onErrorResumeNext(t -> { 
               System.out.println("Error after all retries:" + t.getCause()); 
               return Observable.just("I save the world for extinction!"); 
              }) 
              .subscribe(s -> System.out.println(s)); 
    new TestSubscriber((Observer) subscription).awaitTerminalEvent(500, TimeUnit.MILLISECONDS); 
} 

Если вы хотите увидеть больше примеров, посмотрите здесь: https://github.com/politrons/reactive