如何延迟重试使用 RxJS5 发送 HTTP 请求？

Question

我在使 RxJS5 可观察流以我希望的方式运行时遇到问题。

该流应该使用 axios 向网站发送 HTTP 请求，如果响应是 HTTP 错误（axios 强制为 JavaScript 错误），则可观察序列应等待 10 毫秒，然后尝试重新发送请求（出于某种原因，当您立即重试发送请求并不断抛出错误时，我向其发送请求的网站不喜欢它，但大多数情况下表现良好，延迟 10 毫秒。

Rx.Observable
  .fromPromise(axios('http://example.com/12345'))
  .map(x => new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve(x)
    }, 2000)
  }))
  .debounceTime(2000)
  .do(console.log)
  .retry(10)
  .subscribe(console.log, console.error)

我有一个关于 Codepen 的示例，其中做了一些更改，以便更清楚地了解流的工作原理：http://codepen.io/leakyabstractions/pen/pNmvyZ?editors=0010

我尝试使用.delay()、.debounceTime()、.timer()、.timeInterval() 和.timeout() 代替.map() 运算符，但没有任何效果（包括.map()）。我做错了什么？

Answer 1

所以基本上你正在寻找的是“10 毫秒后重试，但只有 10 次”？ （这就是你的retry(10) 所建议的。 我认为一个复杂的解决方案将在这里包括retryWhen：

const mockedRestError$ = Rx.Observable.throw("http://example.com/12345");

// we'll start with an empty string, because otherwhise
// we could not log the "start..."
Rx.Observable.of("")
  .do(() => console.log("start..."))
  .switchMapTo(mockedRestError$)
  .retryWhen(tenTimesWithDelay)
  .subscribe(console.log, console.error, console.info); // is never called, because 


function tenTimesWithDelay(errors) {
  return errors
    .scan((count, err) => {
      ++count;
      // optionally to throw the error all the way down to the subscription
      // comment the throw out to have the stream simply complete
      if (count >= 10) {
        throw err;
      }
      return count;
    }, 0)
    .takeWhile(count => count < 10)
    .do(count => console.log(`Retry #${count} in 100ms...`))
    .delay(100);
}

这里是代码笔：http://codepen.io/anon/pen/bBydwZ?editors=0010

还请注意，我将延迟设置为 100 毫秒而不是 10 毫秒，这样它在控制台中显示得更清晰一些。

Answer 2

olsn 的回答有效，但是我想分享另一个我不小心想到的解决方案，在我看来这更直接：

console.log('start')
Rx.Observable
// emit observable every 10 ms, change to a bigger number to observe results
.interval(10)
// start immediately
.startWith(0)
// do it 10 times
.take(10)
.do(x => console.log('emitting', x))
// for every observable emitted, do an HTTP request
.flatMap(() => new Promise((resolve, reject) => resolve('resolved promise')))
.first(x => !(x instanceof Error))
.subscribe(console.log, console.warn, () => console.info('done'))