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(...);