repeatWhen（）の動的遅延値

Question

今、私はRxJavaでポーリングロジックを実装しています。私は停止するように指示されるまで、エンドポイントを何度もポーリングすることになっています。さらに、それぞれの応答は、私が再びポーリングする前に遅延するはずの時間とともに戻ってきます。私の論理は今、この権利のようになります。

service.pollEndpoint() 
    .repeatWhen(observable -> observable.delay(5000, TimeUnit.MILLISECONDS)) 
    .takeUntil(Blah::shouldStopPolling); 

今私は、遅延値が5000にハードコード持っているが、私はそれがポーリング応答の値に依存したいと思います。私はObservable.just(pollResponse).repeatWhen(observable -> observable.delay(pollResponse.getDelay(), TimeUnit.MILLISECONDS))を返すフラットマップを使ってみましたが、ソースObservableを混乱させてしまったので、それは正しいアイデアのようには見えませんでした。私はそれが私が見落としている単純なもののように感じる。ありがとう！

Answer 1

あなたが遅延変数を更新するために、副作用演算子doOnNextを使用して、@JohnWowUsが述べたように

int pollDelay = 5000; 

service.pollEndpoint() 
.doOnNext(pollResponse -> pollDelay=pollResponse.getDelay()) 
.repeatWhen(observable -> observable.delay(pollDelay, TimeUnit.MILLISECONDS)) 
.takeUntil(Blah::shouldStopPolling); 

Answer 2

あなたrepeatWhenに、あなたはアウト・オブ・バンド通信が必要ですが、あなたが購読している場合ことを使用することができます

Observable.defer(() -> { 
    int[] pollDelay = { 0 }; 
    return service.pollEndpoint() 
    .doOnNext(response -> pollDelay[0] = response.getDelay()) 
    .repeatWhen(o -> o.flatMap(v -> Observable.timer(pollDelay[0], MILLISECONDS))) 
    .takeUntil(Blah::shouldStopPolling); 
}); 

Answer 3

これは私が使用して終了ソリューションは、次のとおりです：シーケンス回以上、あなたがdeferを使用することができますが、加入者ごとの状態を持っている

public static Observable<PollResponse> createPollObservable(RetrofitService service, PollResponse response) { 
    return Blah::shouldStopPolling(response) 
     ? Observable.empty() 
     : service 
      .pollEndpoint() 
      .delaySubscription(getPollDelay(response), TimeUnit.MILLISECONDS) 
      .concatMap(response1 -> createPollObservable(service, response1) 
        .startWith(response1) 
        .takeUntil(Blah::shouldStopPolling) 
      ); 
} 

代わりに再帰を使用して常に最新のPollResponseオブジェクトを持ち、さらにrepeatWhen()ではなくdelaySubscription()に切り替えます。

Answer 4

あなたでし虐待retryWhen - しかし、私はちょうどそれはあなたがそれやるべきではないということは可能であると言っている：

package com.example.retrywhen; 

import com.example.LoggingAction1; 

import org.pcollections.PVector; 
import org.pcollections.TreePVector; 

import java.util.concurrent.TimeUnit; 
import java.util.concurrent.atomic.AtomicInteger; 

import rx.Observable; 
import rx.functions.Action0; 
import rx.functions.Func0; 
import rx.functions.Func1; 
import rx.schedulers.Schedulers; 

import static com.example.Utils.sleep; 

public class RetryWhenDynamicDelayTest { 

    final static PVector<Integer> delays = TreePVector.<Integer>singleton(500).plus(1_000).plus(2_000); 

    final static AtomicInteger count = new AtomicInteger(0); 

    final static Observable<Integer> willCycleThroughTheList = Observable.defer(new Func0<Observable<Integer>>() { 
     @Override 
     public Observable<Integer> call() { 
      return Observable.just(delays.get(count.getAndIncrement() % 3)); 
     } 
    }); 

    static class ThisIsNotReallyAnException extends Throwable { 
     final Integer integer; 

     ThisIsNotReallyAnException(Integer integer) { 
      this.integer = integer; 
     } 
    } 


    public static void main(String[] args) { 

     final long now = System.currentTimeMillis(); 

     willCycleThroughTheList.flatMap(new Func1<Integer, Observable<?>>() { 
      @Override 
      public Observable<?> call(Integer integer) { 
       return Observable.error(new ThisIsNotReallyAnException(integer)); 
      } 
     }) 
       .doOnUnsubscribe(new Action0() { 
        @Override 
        public void call() { 
         System.out.println("Millis since start: " + (System.currentTimeMillis() - now)); 
        } 
       }) 
       .onErrorResumeNext(new Func1<Throwable, Observable<Integer>>() { 
        @Override 
        public Observable<Integer> call(Throwable throwable) { 
         if (throwable instanceof ThisIsNotReallyAnException) { 
          ThisIsNotReallyAnException thisIsNotReallyAnException = (ThisIsNotReallyAnException) throwable; 
          return Observable.just((thisIsNotReallyAnException.integer)).concatWith(Observable.error(throwable)); 
         } else { 
          return Observable.error(throwable); 
         } 
        } 
       }) 
       .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() { 
        @Override 
        public Observable<?> call(Observable<? extends Throwable> observable) { 
         return observable.flatMap(new Func1<Throwable, Observable<?>>() { 
          @Override 
          public Observable<?> call(Throwable throwable) { 
           if (throwable instanceof ThisIsNotReallyAnException) { 
            ThisIsNotReallyAnException thisIsNotReallyAnException = (ThisIsNotReallyAnException) throwable; 
            return Observable.timer(thisIsNotReallyAnException.integer, TimeUnit.MILLISECONDS); 
           } else { 
            return Observable.error(throwable); 
           } 
          } 
         }); 
        } 
       }) 
       .subscribeOn(Schedulers.io()) 
       .subscribe(new LoggingAction1<Object>("")); 

     sleep(10_000); 
    } 
} 

プリント：

Millis since start: 75 
call(): 500 
Millis since start: 590 
call(): 1000 
Millis since start: 1591 
call(): 2000 
Millis since start: 3593 
call(): 500 
Millis since start: 4094 
call(): 1000 
Millis since start: 5095 
call(): 2000 
Millis since start: 7096 
call(): 500 
Millis since start: 7597 
call(): 1000 
Millis since start: 8598 
call(): 2000