RxJava有条件地延迟响应

Question

我想将响应延迟最多500毫秒。如果updateData已经说了400毫秒，那么我想延迟100毫秒。

使用以下代码总是我延迟500毫秒。目前如果后端需要600毫秒，那么总延迟是1100毫秒，但如果它超过500毫秒我不想延迟。

compositeDisposable.add(someRepo.updateData(dataRequest)
.observeOn(schedulerProvider.mainThread())
.delay(500, TimeUnit.MILLISECONDS)
.subscribeWith(dataUpdateObserver())

Answer 1

试试这样吧。当你启动someRepo.updateData(dataRequest)时启动计时器parellally。然后在完成someRepo.updateData(dataRequest)时获取计时器计数并找到所需的延迟。

如果它已经超过500毫秒，则不需要进一步的延迟。否则找到所需的延迟（500 - elapsedTime）。以下代码也是如此。在这里someRepo.updateData(dataRequest)返回Single<DataResponse>

   Subject<Boolean> timerStopper = PublishSubject.create(); // To terminate the timer
   Observable<Long> timer = Observable.interval(1, TimeUnit.MILLISECONDS)
                .takeUntil(timerStopper).publish().refCount();


   someRepo.updateData(dataRequest).doOnSubscribe(__ -> timer.subscribe())
  .flatMap(dataResponse -> timer.flatMapSingle(elapsed -> {
         timerStopper.onNext(true); // take the elapsed time and terminate the timer
         long requiredDelay = elapsed > MAX_DELAY ? 0 : MAX_DELAY - elapsed; // find the required delay from elapsed. If its already exceeded Maximum delay, no delay is required
         Log.d(TAG, "requiredDelay: " + requiredDelay);
         return Single.fromCallable(() -> dataResponse).delay(requiredDelay, TimeUnit.MILLISECONDS);})
        .map(data -> data).firstOrError())
   .subscribeWith(dataUpdateObserver());

Answer 2

您可以定义一个全局变量来获取updateData操作的总时间，并根据花费的时间来使用延迟值

long total = 0;

@Test
public void customDelay() {
    long start = System.currentTimeMillis();
    Subscription subscription = Observable.just("hello reactive world with custom delay")
            .map(value -> {
                //updateData logic
                try {
                    Thread.sleep(new Random().nextInt(600));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                total = System.currentTimeMillis() - start;
                if (total > 500) total = 0;
                return value;
            })
            .delay(total, TimeUnit.MILLISECONDS)
            .subscribe(n -> System.out.println("time:" + (System.currentTimeMillis() - start)));
    new TestSubscriber((Observer) subscription).awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
}

你可以在这里看到运行https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/utils/ObservableDelay.java的例子