RxJS unsubscribe from inner observable when the source completes

Question

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

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

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

backtick backtick · Accepted Answer · 2020-04-27T22:07:26

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 unsubscribe from inner observable when the source completes

1 Answers