可观察管道中的异常被抑制

Question

假设我希望 observable 定期发出值，直到另一个 observable 发出。所以我可以使用timer 和takeUntil 来实现这一点。 但是然后我想处理每个发出的值并在某些条件变为真时停止（错误）发出。所以我写下一段代码：

const { timer, Subject } = 'rxjs';
const { takeUntil, map } = 'rxjs/operators';

const s = new Subject();
let count = 0;
function processItem(item) {
  count++;
  return count < 3;
}

const value = "MyValue";

timer(0, 1000).pipe(
      takeUntil(s),
      map(() => {
        if (processItem(value)) {
          console.log(`Processing done`);
          return value;
        } else {
          s.next(true);
          s.complete();
          console.error(`Processing error`);
          throw new Error(`Stop pipe`);
        }
      }),
    )

Playground

但我没有收到错误，而是完成了我的 Observable。 仅当我注释掉 takeUntil(s) 运算符时，我才会收到错误消息。 看起来当管道运算符完成时，它的值不会立即发出，而是在管道的下一次“迭代”结束时被记住并发出，然后被新结果替换，依此类推。在我的情况下，下一次迭代，当应该发出错误时，takeUntil 会阻止它。问题是我对这个假设是否正确，如果是，为什么 rxjs 是这样设计的？

Answer 1

首先，每个 Rx 链可以发出一个 error 或一个 complete 通知，但不能同时发出两者。 （参见http://reactivex.io/documentation/contract.html 部分“管理通知的合同”）。

takeUntil 运算符发出 complete 当它是通知 Observable（s 在你的情况下）发出任何 next 通知。这意味着当s 发出时，链将完成，您将永远不会收到任何进一步的error 通知。

最后一件事，可能也是最令人困惑的是，RxJS 中的所有事情都是同步发生的，除非您使用时间（例如 delay 运算符）或者您专门将 observeOn 运算符与异步调度程序一起使用。因此，当您在map 中调用s.next(true) 时，此next 通知会立即传播到完成链的takeUntil，正如我上面提到的，您可以收到一个error 或一个complete 通知，但不能同时收到两者。

看起来您甚至不需要使用 takeUntil，因为如果您在 map 内抛出错误，它会自动包装并作为 error 通知 (How to throw error from RxJS map operator (angular)) 进一步发送，并且链已被处置自动，因此在此之后尝试使用 takeUntil 完成它是没有意义的。