javascript - RxJS:takeUntil 忽略主题事件

Question

在下面的代码示例中，目的是通过向 Subject mid$ 发出 1 来阻止来自 second$ 的事件。

import { Subject, timer } from "rxjs";
import { switchMap, takeUntil, tap } from "rxjs/operators";

const first$ = timer(1000);
const second$ = timer(2000);

const mid$ = new Subject();

first$.pipe(
  tap(() => { 
    mid$.next(1); 
  }),
  switchMap(() => second$.pipe(
    takeUntil(mid$),
    tap(() => console.log("MISSED!"))
  )),
).subscribe();


mid$.subscribe(() => console.log("RECEIVED"));

Stackblitz

但由于某种原因它不起作用，如控制台所示:

RECEIVED
MISSED!

即mid$.next(1); 行中的事件发射未被 takeUntil(mid$)

考虑

这里的逻辑是什么？

我注意到，如果我将行 mid$.next(1); 替换为 timer(0).subscribe(() => mid$.next(1)); 它按预期工作，但我想知道在 RxJS 中处理此类情况的正确方法是什么。

Answer 1

它没有像预期的那样工作

const first$ = timer(1000);
const second$ = timer(2000);

const mid$ = new Subject();

first$.pipe(
  tap(() => { 
    mid$.next(1);
  }),
  switchMap(() => second$.pipe(
    takeUntil(mid$),
    tap(() => console.log("MISSED!"))
  )),
).subscribe();

因为当到达 mid$.next(1); 时，尚未创建 switchMap 的内部可观察对象。所以，takeUntil 还没有订阅那个 mid$ 主题。

它适用于 timer(0).subscribe(() => mid$.next(1));(与 setTimeout(() => mid 大致相同$.next() , 0))，因为在这种情况下，当 mid$ 发出时，switchMap 已经创建了内部可观察对象。

解决此问题的快速方法可能涉及使用 BehaviorSubject 而不是 Subject，因为 BehaviorSubject 会将最后发出的值发送到新订阅者:

const first$ = timer(1000);
const second$ = timer(2000);

const mid$ = new BehaviorSubject(null);

first$.pipe(
  tap(() => { 
    mid$.next(1);
  }),
  switchMap(() => second$.pipe(
    // here, when `mid$` is subscribed, the subscriber will receive `1` 
    // and the entire inner observable will complete
    takeUntil(mid$),
    tap(() => console.log("MISSED!"))
  )),
).subscribe();