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

Question

我知道这是因为这里desribed

• 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" 

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

可观察的解决方案是使用.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); 
} 

想法？

P.S.

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

就像任何人提到这个问题的附加信息一样。

Answer 1

在你只需要一个单独的参考Subject以及对你追加observeOn操作后，链的引用情况：

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

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