RXJS：如何使用 takeUntil 直到另一个 Observable 完成（并且不发出）

Question

我有两个可观察对象：

1：累加一个值，完成。 （重要提示：可能在发出任何值之前完成！）

2： 执行操作直到 1 完成（不发出）

你会如何在 RxJS 中实现这个？

我的第一次尝试是使用last()，例如：

firstObservable.pipe(
    takeUntil(secondObservable.pipe(last()))
  ).subscribe(count =>{
    //do stuff with count
  });

但是当firstObservable完成而没有发出任何值时，问题就出现了。 出现错误：

Error: no elements in sequence

问题：如何使用 takeUntil （或任何其他运算符）在 observable 2 完成后停止订阅，并且在没有发出最后一个值时不抛出错误？

Answer 1

您可以在

secondOvservable

上使用 Finalize 并完成那里的

firstObservable

。这样它不依赖于发出的任何值，只依赖于完整的事件。

secondOvservable.pipe(
  finalize(() => firstObservable.complete())
).subscribe(count =>{
  //do stuff with count
});

Answer 2

我一直在使用一种我发现对于任意事件触发

takeUntil()

有用的模式。您基本上创建了您直接控制的第二个可观察的“terminateSignal$”。

const terminateSignal$ = new Subject<void>();
secondObservable.subscribe({
  complete: () => terminateSignal$.next()
});

firstObservable
  .pipe(takeUntil(terminateSignal$))
  .subscribe( /* do stuff */ );

Answer 3

具体化可观察对象以接收类型的通知：下一步、错误、完成。然后简单地过滤掉它们。

 takeUntil(yourObs$.pipe(materialize(), filter(x=> x.kind === 'C'))))