RxJs : Weird behavior with takeUntil?

Question

这里没什么太严肃的，只是好奇。
我想举个例子，想出了这段代码:

const { Observable, Subject } = Rx

const timeout$ = new Subject()

const obs$ = Observable
  .of(1)
  .takeUntil(timeout$)
  .delay(2000)
  .subscribe(x => console.log(x))

timeout$.next()
timeout$.complete()

我原以为这段代码不会显示console.log，但它确实显示了。

有人能解释一下为什么吗？
这是错误还是我对 takeUntil 的理解不正确？

这是一个用于演示的 Plunkr https://plnkr.co/edit/wpKztBabnBeIuNZS28wu?p=info

Answer 1

请注意，如果您切换 takeUntil() 和 delay() 的顺序，它不会按预期发出:

Observable
  .of(1)
  .delay(2000)
  .takeUntil(timeout$)
  .subscribe(x => console.log(x));

你看到的是因为 delay() 在 takeUntil() 发送 complete 通知之前安排了发射。
现在在 RxJS 5.2.0 中，takeUntil() 运算符不会强制取消订阅它的源，这就是问题所在。因此，当 takeUntil() 被 timeout$ 通知时，它仍然订阅了它的源 Observable，因此来自 delay() 的计划发射是'不处理。

注意在处理来自通知 Observable 的发射时没有 this.unsubscribe() 调用:https://github.com/ReactiveX/rxjs/blob/master/src/operator/takeUntil.ts#L61

还有其他运算符具有相同的行为。例如 first() 和 takeWhile()。

另一方面，例如 take() 运算符取消订阅:

https://github.com/ReactiveX/rxjs/blob/master/src/operator/take.ts#L80

此行为实际上已被报告，现在正在讨论这是否是错误:

• https://github.com/ReactiveX/rxjs/issues/2455

• https://github.com/ReactiveX/rxjs/pull/2470

• https://github.com/ReactiveX/rxjs/pull/2463

• https://github.com/ReactiveX/rxjs/pull/2457