当源完成时，RxJS 取消订阅内部 observable [英] RxJS unsubscribe from inner observable when the source completes

问题描述

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

I have a function that returns observable that automatically unsubscribe when some internal thing happens using a Subject. Here is a simple version that shows the issue:

const subject = new Subject();

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

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

问题是，如果有人订阅了 source 并使用了高阶 observable，当 Subject 发出时，它不会从内部 observable 订阅:

The issue is that if someone subscribes source and uses higher-order observable, it will not subscribe from the inner observable when the Subject emits:

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

有没有办法在不强制消费者调用unsubscribe的情况下解决这个问题?

There is a way to solve this issue without forcing the consumer to call unsubscribe?

推荐答案

问题是 source 完成并不意味着当前运行间隔结束.如果您想在源关闭时终止内部可观察对象，您可以这样做，您只需要多播 source 以避免创建更多 h1 事件侦听器.

The issue is that source completing doesn't mean the currently running interval(s) end. If you want to kill the inner observables thing when source closes, you can do that, you'll just want to multicast source to avoid creating more h1 event listeners.

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)
);

这篇关于当源完成时，RxJS 取消订阅内部 observable的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！