RxJS takeUntil не отписывается

Question

Хочу отписаться в декларативном стиле с оператором takeUntil. Но это в принципе не работает. В любом случае, я вижу вывод консоли.

const unsubscribe = new Subject();

function printFoo() {
  of('foo')
    .pipe(takeUntil(unsubscribe))
    .subscribe(console.log) // Why I can see 'foo' in the  console?
}

function onDestroy() {
  unsubscribe.next();
  unsubscribe.complete();
}

onDestroy()
setTimeout(() => printFoo(), 200)

StaackBlitz:

https://stackblitz.com/edit/rxjs-svfkxg?file=index.ts

P.S. Я ожидал, что даже unsubscribe.next() будет достаточно, чтобы отписаться, но даже с unsubscribe.complete() это не сработает.

Answer 1

Вы звоните onDestroy() до того, как цепочка с takeUntil будет даже создана.

Когда вы в конечном итоге назовете printFoo(), предыдущие выбросы на unsubscribe не будут переизданы, и субъект unsubscribe все равно уже завершен, поэтому takeUntil в этом случае никогда не завершит цепочку.

Answer 2

Поскольку субъект отправляет до подписки printFoo.

После подписки больше нет выделений субъекта.

Вместо этого вы можете использовать BehaviorSubject, поскольку он содержит испущенные значения (последниеиспускаемое значение):

const unsubscribe = new BehaviorSubject(false);

function printFoo() {
  of('foo')
    .pipe(takeUntil(unsubscribe.pipe(filter(value => !!value)))) // Don't unsub if it's false emitted
    .subscribe(console.log)
}

function onDestroy() {
  unsubscribe2.next(true); // Emit true to cancel subscription
}