rxjs - 在 .switchMap 运算符中返回 .empty Observable

Question

我想我在 RxJS 库中发现了一个错误，或者只是无法理解它的行为。看看下面的代码:

const { Observable, Subject } = Rx;

const source$ = new Subject();

const handle = v => {
  if (v === 0) {
    source$.next(1);
    return Observable.empty();
  }  
  return Observable.interval(1000).startWith(-1).take(3);
}

source$
  .do(e => console.log('source:', e))
  // .delay(1)
  // .switchMap(v => Observable.of(null).switchMap(() => handle(v)))
  .switchMap(v => handle(v))
  .subscribe(e => console.log('handle:', e))

source$.next(0); // <- initial value

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

工作场景: 当我使用 source$.next(1) 运行它时，我看到以下结果(正确):

source: 1
handle: -1
handle: 0
handle: 1

错误场景:当我使用 source$.next(0) 运行它时，我希望看到:

source: 0
... (exactly the same output as for the previous example)

但我只看到 3 个值:

source: 0
source: 1
handle: -1

解决方法 1: 当我像您在评论中看到的那样放置 .delay(1) 时 - 它工作正常，正如我预期的那样。我认为这是因为默认调度程序在同一时间同步执行，而不是在使用延迟时使用异步调度程序。

问题 当我使用 Observable.of(null) (请参阅代码中的注释)时，它也会给我相同的(“正确的”或预期的)结果。这是为什么？不应该和之前一样同步吗？

Answer 1

行为是正确的。这是来自 source 的同步发射这让它有点奇怪。

发射 1来自 handle方法会影响一个重入调用，它会看到 startWith -1 的值在 Observable.empty 之前 发出返回 - 作为 startWith值同步发出。

这意味着 switchMap看到 startWith/interval首先可观察到 empty可观察的第二。然后切换到 empty可观察的，影响你看到的输出。

添加 delay防止重新进入调用。