当源完成时，RxJS 取消订阅内部 observable

Question

我有一个返回 observable 的函数，当使用 Subject 发生一些内部事情时，它会自动取消订阅。这是一个显示问题的简单版本：

const subject = new Subject();

const source = fromEvent(document.querySelector("h1"), "click").pipe(
  takeUntil(subject)
);

fromEvent(document.querySelector("p"), "click").subscribe(() => {
  subject.next();
});

问题在于，如果有人订阅了source 并使用了高阶可观察对象，那么当Subject 发出时，它将不会订阅内部可观察对象：

source.pipe(mergeMap(() => interval(1000))).subscribe(x => console.log(x));

有一种方法可以解决这个问题，而无需强制消费者致电unsubscribe？

Answer 1

问题是source 完成并不意味着当前运行的间隔结束。如果你想在源关闭时杀死内部的 observables，你可以这样做，你只需要多播 source 以避免创建更多的 h1 事件监听器。

const subject = new Subject();

const source = fromEvent(document.querySelector("h1"), "click").pipe(
  takeUntil(subject)
);

const sharedSource = source.pipe(share());
const emitOnComplete = last(null, null);
const sourceIsDone = sharedSource.pipe(emitOnComplete);

sharedSource.pipe(
  mergeMap(() => interval(1000)),
  takeUntil(sourceIsDone)
);