增加流的延迟？

Question

我有以下内容，它确实有效，它不断增加延迟并最终超时，这正是我想要的。

但是因为我使用的是 Concatmap，所以我失去了区间的原始值。

   let x = 1
    let source2$ = interval(500)
      .pipe(

        concatMap(() => {
          x++
          let newtime = x * 500
          console.log("newtime ", newtime)
          return of(5).pipe(delay(newtime))
        }),
        timeout(3000),
        map((data) => {
          return 'Source 2: ' + data
        })
      )

所以它打印 Source 2: 5.. 我希望它打印间隔的值。

我使用 concatmap 完成了我想要的工作，但我认为它是错误的运算符，因为我失去了原始值。

有人可以帮忙吗？

更多信息

总而言之，我想做的就是使用间隔发出值，并在每次发出后增加延迟时间 - 最终达到 3000 毫秒的超时并出错。

Answer 1

我在 cmets 中提到，您可以使用 concatMap 来接收来自 interval 的不断增加的索引：

concatMap(index => {
  let newtime = index * 500
  console.log("newtime ", newtime)
  return of(index).pipe(delay(newtime))
}),

请注意，我将通过 of(index) 将值返回到流中。

我想我明白你对返回另一个 Observable 有什么顾虑。由于您想按顺序发出项目（仅在前一个完成后才发出一个），因此您必须将 concatMap 与另一个内部 Observable 一起使用。没有专门针对此功能的运算符，因为这是“可组合行为”，这意味着您可以通过组合现有运算符来实现此行为。

Answer 2

const source2$ = interval(500)
      .pipe(
        map(x => x * 500),
        switchMap(x => timer(x)),
        timeout(3000),
        map(data =>  'Source 2: ' + data)
       )

更新： https://stackblitz.com/edit/rxjs-iywcm6?devtoolsheight=60

const source2$ = interval(500)
      .pipe(
        tap(x => console.log('Tick before delay', x)),
        concatMap(x => timer((x + 1) * 500).pipe(mapTo(x))),        
        tap(x => console.log('Tick after delay', x)),
        map(data =>  'Source 2: ' + data),
        timeout(3000)
       ).subscribe(
         (data) => console.log(data),
         e => console.error('Timeout', e))