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

Question

我希望轮询端点不快于每秒一次，并且不慢于轮询端点所需的时间。决不能有超过一个未完成的请求。

我想要一种响应式编程方式来至少每秒轮询一次端点，但如果端点花费的时间超过 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 周期慢时不会累积请求。

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

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>

Answer 2

我相信这可以满足您的需求：

let counter = 0;
function apiCall() {
  const delay = Math.random() * 1000;
  const count = ++counter;
  return Rx.Observable.timer(delay).mapTo(count);
}

Rx.Observable.timer(0, 1000)
  .mergeMap(() => apiCall())
  .take(1)
  .repeat()
  .subscribe(x => { console.log(x); });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.js"></script>

• timer(0, 1000)：立即发射，之后每隔一秒发射一次
• mergeMap(...)：切换到 api 调用返回的 observable。这将在每次重试时生成一个新的 observable。如果您不想在每次重试时创建一个新的，请将其替换为 mergeMapTo(apiCall())。
• take(1)：强制订阅完成，因此一旦 api 发出，计时器就不会触发
• repeat()：在 api 发出时重新开始序列

所以调用将立即调用 api。如果它在一秒钟内没有返回，则每秒将进行另一个调用。一旦有来自 api 调用之一的响应，计时器将被取消，整个序列重新开始。这不会取消我认为符合您意图的正在进行的请求。

编辑：如果后一个请求在前一个请求之前返回，那么前一个请求将被丢弃。

Answer 3

在我想出一个仅基于 rxjs 且没有副作用（没有变量分配）且没有背压的答案之前，我确实必须考虑 1500 万！

const { Observable } = Rx;

const mockHttpRequest = url =>
  Observable
    .of('ok')
    .do(x => console.log('fetching...'))
    .delay(250);

const poll = (httpRequest$, ms) => {
  const tick$ = Observable.timer(ms);

  return Observable
    .zip(httpRequest$, tick$)
    .repeat()
    .map(([httpResult]) => httpResult);
};

poll(mockHttpRequest('your-url-here'), 1000)
  .do(console.log)
  .subscribe();

这是一个正常工作的 Plunkr：https://plnkr.co/edit/sZTjLedNCE64bgLNhnaS?p=preview