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请求才能重述，因为这是新的可观察到的。必须有某种方法将链的开头封装到一个可观察的单个？

Answer 1

请参阅Learnrxjs -Renry。该示例显示了在丢弃错误时从源开始重新启动的所有内容。

页面显示pipe语法，但是如果您喜欢

基本模式是

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

Answer 2

简单的方法是将您的复合物在延期添加和可观察到的可观察到的复合物中使用。