RxJs - concatMap 替代方案，可以删除介于两者之间的所有内容

Question

我试图找到一个行为类似于 concatMap 的运算符，但会丢弃介于两者之间的所有内容。例如， concatMap 执行以下操作:

• next a
• start handling a
• next b
• next c
• finish handling a
• start handling b
• finish handling b
• start handling c
• finish handling c

相反，我正在寻找一种会掉落的机制 b , 自 c已经进来了:

• next a
• start handling a
• next b
• next c
• finish handling a
• (skip b)
• start handling c
• finish handling c

有关更扩展的示例，请参阅此要点: https://gist.github.com/Burgov/afeada0d8aad58a9592aef6f3fc98543

Answer 1

我认为您要找的接线员是 throttle .

这是一个工作Stackblitz .完成这项工作的关键是设置传递给 throttle() 的配置对象。这允许它排放(和处理)前导和尾随源排放，但忽略两者之间的任何排放，在 processData() 期间在跑。

这是 Stackblitz 的关键功能:

// Use 'from' to emit the above array one object at a time
const source$ = from(sourceData).pipe(

  // Simulate a delay of 'delay * delayInterval' between emissions
  concatMap(data => of(data).pipe(delay(data.delay * delayInterval))),

  // Now tap in order to show the emissions on the console.
  tap(data => console.log('next ', data.emit)),

  // Finally, throttle as long as 'processData' is handling the emission
  throttle(data => processData(data), { leading: true, trailing: true }),

).subscribe()

简短而甜蜜，除了一个问题外，按要求工作......

更新:

上面代码的“一个问题”是当源 observable 完成时， throttle()取消订阅 processData，有效地停止需要完成的任何最终处理。正如 Bart van den Burg 在下面的评论中指出的那样，修复方法是使用主题。我认为有很多方法可以做到这一点，但 Stackblitz 已更新为以下代码，现在可以使用:

// Set up a Subject to be the source of data so we can manually complete it
const source$ = new Subject();

// the data observable is set up just to emit as per the gist.
const dataSubscribe = from(sourceData).pipe(
    // Simulate a delay of 'delay * delayInterval' before the emission
    concatMap(data => of(data).pipe(delay(data.delay * delayInterval))),
).subscribe(data => {
    console.log('next ', data.emit); // log the emission to console
    source$.next(data); // Send this emission into the source
});

// Finally, subscribe to the source$ so we can process the data
const sourceSubscribe = source$.pipe(
    // throttle as long as 'processData' is handling the emission
    throttle(data => processData(data), { leading: true, trailing: true })
).subscribe(); // will need to manually unsubscribe later ...