如何實現開關的IObservable <IObserver<T>>在C＃中

Question

在反應擴展我們

IObservable<T> Switch(this IObservable<IObservable<T>> This) 

反應擴展我想的

IObserver<T> Switch(this IObservable<IObserver<T>> This) 

實現這將切換傳出事件不同的觀察者但 作爲一個單一的觀察員呈現。

Answer 1

此版本處理幾個問題：

• 有可能導致丟失事件的競爭條件。如果觀察者在一個線程上觀察事件，而源觀察者在另一個線程上產生一個新的觀察者，如果不使用任何類型的同步，則可能最終在其他線程調用之前在一個線程上對當前觀察者調用OnCompletedOnNext在同一觀察員。這將導致事件丟失。

• 與上面相關，默認情況下，觀察者不是線程安全的。您不應該同時打電話給觀察員，否則您將違反主要Rx合同。在沒有任何鎖定的情況下，用戶可能會在currentObserver上調用OnCompleted，而另一個線程則在該觀察者上調用OnNext。開箱即用，這種事情可以通過使用Synchronized Subject來解決。但是因爲我們還需要同步以前的問題，所以我們可以使用簡單的互斥鎖。

• 我們需要一種方法來取消訂閱源observable。我認爲，當結果觀察者完成（或錯誤）時，由於我們的觀察者已經被告知不再需要事件，所以這是退訂源的好時機。

下面的代碼：

public static IObserver<T> Switch<T>(this IObservable<IObserver<T>> source) 
{ 
    var mutex = new object(); 
    var current = Observer.Create<T>(x => {}); 
    var subscription = source.Subscribe(o => 
    { 
     lock (mutex) 
     { 
      current.OnCompleted(); 
      current = o; 
     } 
    }); 

    return Observer.Create<T>(
     onNext: v => 
     { 
      lock(mutex) 
      {     
       current.OnNext(v); 
      } 
     }, 
     onCompleted:() => 
     { 
      subscription.Dispose(); 
      lock (mutex) 
      { 
       current.OnCompleted(); 
      } 
     }, 
     onError: e => 
     { 
      subscription.Dispose(); 
      lock (mutex) 
      { 
       current.OnError(e); 
      } 
     }); 
} 

Answer 2

public static IObserver<T> Switch<T>(this IObservable<IObserver<T>> This) 
{ 
    IObserver<T> currentObserver = Observer.Create<T>(x => { }); 

    This.Subscribe(o => { currentObserver.OnCompleted(); currentObserver = o; }); 


    return Observer.Create<T> 
     (onNext: v => currentObserver.OnNext(v) 
     , onCompleted:() => currentObserver.OnCompleted() 
     , onError: v => currentObserver.OnError(v)); 
}