rxjs 定期轮询具有可变响应时间的端点

Question

我希望轮询端点的速度不超过每秒一次，并且不慢于轮询端点所需的时间。永远不应该有超过一个未决请求。
我想要一种响应式(Reactive)编程方式，至少每秒轮询一次端点，但如果端点花费的时间超过 1 秒，则立即触发下一个请求。
在下面的弹珠图中，第 2 和第 3 个请求花费的时间超过 1 秒，但第 4 和第 5 个请求完成得更快。下一个请求在 1 秒边界上触发，或者在从最后一个未完成的请求中获取数据后立即触发。

s---s---s---s---s---s---| # 1 second interval observable
r---r----r--------r-r---| # endpoint begin polling events
-d-------d--------dd-d--| # endpoint data response events

我试图在大理石图中获得正确的术语，所以我
假设端点请求的开始应该是
我标记为“r”的弹珠，标记为“d”的弹珠事件具有端点数据。
这是我在纯 js 中执行此操作需要多少代码；但是，如我上面所要求的那样，后续请求不会在获得后立即触发。

var poll;
var previousData;
var isPolling = false;
var dashboardUrl = 'gui/metrics/dashboard';
var intervalMs = updateServiceConfig.getIntervalInMilliSecondForCharts();

return {
    startInterval: startInterval,
    stopInterval: stopInterval
};

function startInterval() {
    stopInterval();
    tryPolling(); // immediately hit the dashboard
    // attempt polling at the interval
    poll = $interval(tryPolling, intervalMs);
}

/**
 * attempt polling as long as there is no in-flight request
 * once the in-flight request completes or fails, allow the next request to be processed
 */
function tryPolling() {
    if (!isPolling) {
        isPolling = true;

        getDashboard()
        // if the dashboard either returns successful or fails, reset the polling boolean
            .then(resetPolling, resetPolling);
    }
}

/** there's no longer an in-flight request, so reset the polling boolean */
function resetPolling() {
    isPolling = false;
}

function stopInterval() {
    if (poll) {
        $interval.cancel(poll);
        poll = undefined;
    }
}

function getDashboard() {
    return restfulService.get(dashboardUrl)
        .then(updateDashboard);
}

function updateDashboard(data) {
    if (!utils.deepEqual(data, previousData)) {
        previousData = angular.copy(data);
        $rootScope.$broadcast('$dashboardLoaded', data);
    }
}

Answer 1

这是我的解决方案。它使用内部主题，combineLatest和 filter如果响应比 timer 到达速度慢，则确保请求不会累积。时期。

评论应该解释它是如何工作的。

const delays = [100, 2000, 100, 3000];
const since = Date.now();
let index = 0;

function mock() {
    return Rx.Observable
    .of("res")
    .do(() => console.log("mock req at ", Date.now() - since, " ms"))
    .delay(delays[index++ % delays.length])
    .do(() => console.log("mock res at ", Date.now() - since, " ms"));
}

function poll() {

  return Rx.Observable.defer(() => {

    // Use defer so that the internal subject is created for each
    // subscription.
    const subject = new Rx.BehaviorSubject({ tick: -1, pending: false });

    return Rx.Observable
    
      // Combine the timer and the subject's state.
      .combineLatest(
        Rx.Observable.timer(0, 1000).do(tick => console.log("tick", tick)),
        subject
      )

      // Filter out combinations in which either a more recent tick
      // has not occurred or a request is pending.
      .filter(([tick, state]) => (tick !== state.tick) && !state.pending)

      // Update the subject's state.
      .do(([tick]) => subject.next({ tick, pending: true }))
      
      // Make the request and use the result selector to combine
      // the tick and the response.
      .mergeMap(([tick]) => mock(), ([tick], resp) => [tick, resp])

      // Update the subject's state.
      .do(([tick]) => subject.next({ tick, pending: false }))
      
      // Map the response.
      .map(([tick, resp]) => resp);
  });
}

poll().take(delays.length).subscribe(r => console.log(r));

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

我突然想到有一个操作符可以做到这一点: exhaustMap .

const delays = [100, 2000, 100, 3000];
const since = Date.now();
let index = 0;

function mock() {
  return Rx.Observable
    .of("res")
    .do(() => console.log("mock req at ", Date.now() - since, " ms"))
    .delay(delays[index++ % delays.length])
    .do(() => console.log("mock res at ", Date.now() - since, " ms"));
}

const poll = Rx.Observable
  .timer(0, 1000)
  .do(tick => console.log("tick", tick))
  .exhaustMap(() => mock());

poll.take(delays.length).subscribe(r => console.log(r));

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>