RxJS - 一个 observable 'A' 不会停止发射从同一个 observable 'A' 传输的 takeUntil

Question

我有一个可观察的，在单击开始按钮后每秒向控制台打印一个数字。单击开始按钮 5 秒后，可观察对象应停止打印。

开始条件 单击开始按钮后的每一秒

结束条件 点击开始按钮五秒后

不起作用的代码

const startClick$ = fromEvent(document.getElementById('start'), 'click');
const stop$ = startClick$.pipe(switchMapTo(interval(5000)));
const printInterval$ = interval(1000).pipe(takeUntil(stop$));
const startPrint$ = startClick$.pipe(switchMapTo(printInterval$));

startPrint$.subscribe(
  value => {
    console.log('new value is:', value);
  },
  () => {
    console.log('there was some error!');
  },
  () => {
    console.log('i have completed');
  }
);

stop$.subscribe(() => {
  console.log('i should stop right now!!');
});

在点击开始按钮五秒后，startPrint$ Observable 不会停止发射。

有效的代码

const endClick$ = fromEvent(document.getElementById('end'), 'click');
const stopWithEndClick$ = endClick$.pipe(switchMapTo(interval(5000)));
const printInterval1$ = interval(1000).pipe(takeUntil(stopWithEndClick$));
const startPrint1$ = startClick$.pipe(switchMapTo(printInterval1$));

startPrint1$.subscribe(
  value => {
    console.log('1: new value is:', value);
  },
  () => {
    console.log('1: there was some error!');
  },
  () => {
    console.log('1: i have completed');
  }
);

stopWithEndClick$.subscribe(() => {
  console.log('1: i should stop right now!!');
});

这里我有另一个按钮end，我会在点击结束按钮 5 秒后等待。在这种情况下，observable 会按预期停止发射。

我怎样才能使第一个案例工作，我在这里犯了任何错误吗？完整的工作代码可以在https://stackblitz.com/edit/take-until-issue?file=index.ts找到。

Answer 1

这是因为 RxJS 运算符中的订阅顺序而发生的。当您第一次单击“开始”按钮时，第一个收到通知的是stop$，但没有人在收听stop$，因此通知不会触发任何内容。您看到"'i should stop right now!!' 只是因为您最后订阅了自己。

因此，最简单的解决方案是将stop$ 链创建为一个 Observable，只有在您订阅后才会开始发射，就像 timer(5000) 一样，它只发射一次然后完成。

const stop$ = timer(5000);

您更新的演示：https://stackblitz.com/edit/take-until-issue-or4okm?file=index.ts

请注意，您可能会得到不同数量的结果，因为 RxJS（通常是 JavaScript）不能保证 5000ms 将完全是 5000ms。这同样适用于1000ms，因此有时事件可能会以不同的顺序触发。如果您只想要 5 个结果，最好使用 take(5)。此外，对于日志记录，最好使用例如 tap(v => console.log(v))，这样您就不会再次订阅。