javascript - RxJS 重试整个链

Question

我从实时流中读取图像并定期选择一批。然后我将它们发送到服务器进行验证。如果任何验证失败，将引发 HTTP 错误。如果发生这种情况，我想获得一批新图像。

this.input.getImages()
  .throttleTime(500)
  .switchMap(image =>

      new Observable<{}>(observer => {
        // Some operation
      })
      .map(i => ({ image, i }))

  ).filter(({ i }) =>  {
    // some filtering
  })
  .map(({ image }) => image)
  .take(6)
  .bufferCount(6)
  .map(images =>  // switch map??
    Observable.fromPromise(this.server.validate(images))
  )
  .retry(2)  // This only retrys the request, I want it to retry the whole chain (to get valid images)
  .subscribe(images => {
      console.log('All done')
    },
    err => {console.log(err)}
  )

我遇到的问题是只有 HTTP 请求被重试，因为那是新的 observable。必须有某种方法将链的开头封装到单个 Observable 中？

Answer 1

请参阅learnrxjs - 重试。该示例显示了在引发错误时从源代码开始重新启动的所有内容。

该页面显示pipe语法，但如果您愿意，JSBin 会显示流畅的运算符语法。

基本模式是

const retryMe = this.input.getImages()
  .flatMap(val => {
    Observable.of(val)
      // more operators
  })
  .retry(2);

Answer 2

简单的方法是将复杂的 observable 包装在defer中，并对结果 observable 使用重试。