我怎样才能延迟 obs 1 直到 obs 2 没有发出 x 秒？

Question

我有一个obs，它的发射应该等到另一个obs停止发射一段时间后，然后发射最新的。

const {Subject, interval} = rxjs;
const {take} = rxjs.operators;

const subject = new Subject(0)

interval(1000)
  .pipe(take(2))
  // detect that subject has recently fired and delay until
  // 1000 after most recent subject emission, then take only latest
  .subscribe(r => console.log(r))

subject.subscribe(r => console.log('subject emission'))

setTimeout(() => {
  subject.next(1);
}, 500)

setTimeout(() => {
  subject.next(2);
}, 1200)

实际结果：

subject emission
0
subject emission
1

想要的结果

subject emission
subject emission
// one second to be sure no immediate emissions happening
1

根据@benshabatnoam 的评论进行编辑。我使用间隔是因为它是一种可以快速获取的工具，但这是一个错误。在我的实际情况下，这是不可预测的。它应该由一个可以随时发射的 obs 表示。因此，实际期望的结果应该是这样的：

// gold obs fires but notices subject is emitting
subject emission
subject emission
// one second to be sure no immediate emissions happening
1
// gold obs fires a few seconds later and sees there are no subject emissions
2

Answer 1

使用combineLatest、debounceTime和map：

const main$ = new Subject(0);

main$.subscribe(() => console.log('subject emission'))


// aux$ could be any observable
const aux$ = interval(1000).pipe(
  take(2)
)


setTimeout(() => main$.next(1), 500);
setTimeout(() => main$.next(2), 1200);

combineLatest(main$, aux$).pipe(
  map(([main, _aux]) => main),
  debounceTime(1000), // define your own delay time here
).subscribe(result => console.log(result))

结果符合预期。

编辑鉴于新评论，withLatestFrom 可能是您正在寻找的运算符：

aux$.pipe(
    withLatestFrom(main$),
    map(([_aux, main]) => main),
    debounceTime(1000),
)

Answer 2

编辑：如果我对你的理解正确的话，这个管道就可以完成这项工作：

this.obs1.pipe(
  debounceTime(1000),
  switchMap(() => this.obs2)
)

此管道中的逻辑是在 obs 触发时等待一秒钟（使用 debounceTime operator) and then switch it to listen to the second observable (using the switchMap 运算符）。

以下是针对您的问题的演示：

Reproducing the problem

Fixing the problem（已更新）

干杯

我怎样才能延迟 obs 1 直到 obs 2 没有发出 x 秒？

How can I delay obs 1 until obs 2 has not emitted for x seconds?

delay

observable

rxjs

debounce