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)时，请并行启动计时器。然后，在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的示例