rxJs 主题，订阅错误会杀死整个流

Question

我知道这是这里描述的已知行为

• https://medium.com/@benlesh/on-the-subject-of-subjects-in-rxjs-2b08b7198b93

基本问题在上面的链接中有描述，但这里是相关代码（取自链接）

// This is going to behave strangely
const source$ = Observable.interval(1000).share();
const mapped$ = source$.map(x => {
  if (x === 1) {
    throw new Error('oops');
  }
  return x;
});
source$.subscribe(x => console.log('A', x));
mapped$.subscribe(x => console.log('B', x));
source$.subscribe(x => console.log('C', x));
// "A" 0
// "B" 0
// "C" 0
// "A" 1
// Uncaught Error: "oops"

订阅中的错误将终止整个源流。

Observable 的解决方案是使用.observeOn(Rx.Scheduler.asap);

我对整个反应式编程还很陌生，我很难将此解决方案应用于我的Subject，因为主题不支持observeOn。

但我需要Subject，因为我需要将新的价值观推向高潮。

如何解决此问题或将observeOn 与Subject 一起使用？

observeOn 返回一个Observable。但我很难将observeOn 与我的Subject 结合起来。 如何使用 observeOn 并且仍然能够将值推送到我的主题？

这是当前代码（简化）

export class MyClass{

    private messages: Subject<Message> = new Subject<Message>();

    dispatchMessage(message: Message) {

     this.messages.next(message);
}

想法？

附言

对于任何使用 angular 的人（比如我），observeOn 可能会产生一些不希望的副作用。 https://github.com/angular/angular/issues/14316

对于任何提出这个问题的人来说，都是额外的信息。

Answer 1

在这种情况下，您只需要在附加 observeOn 运算符后单独引用 Subject 以及对链的引用：

const subject$ = new Subject();
const obs$ = subject$.observeOn(...);

obs$.subscribe(...);
subject$.next(...);