Как реализовать переключатель для IObservable <IObserver<T>> для реактивных расширений в C#

Question

В реактивных расширениях мы имеем

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

Я хотел бы реализацию

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

, который будет переключать исходящие события в различные наблюдатель но представлен как один наблюдатель.

Answer 1

Эта версия поддерживает несколько вопросов:

• Там в состояние гонки, что может привести к потере событий. Если наблюдатель наблюдает за событием в одном потоке, в то время как наблюдаемый источник создает нового наблюдателя в другом потоке, если вы не используете какую-либо синхронизацию, вы можете в конечном итоге вызвать OnCompleted на текущего наблюдателя в одном потоке непосредственно перед тем, как другой поток вызовет OnNext на том же наблюдателе. Это приведет к потере события.

• Относительно вышесказанного, по умолчанию наблюдатели не являются потокобезопасными. Вы никогда не должны одновременно обращаться к наблюдателю, иначе вы нарушите первичный контракт Rx. Без блокировки абонент может вызвать OnCompleted на currentObserver, а другой поток вызывает OnNext на том же самом наблюдателе. Из-за этого подобные вещи можно решить, используя синхронизированный объект. Но так как нам нужна синхронизация и для предыдущей проблемы, мы можем просто использовать простой мьютекс.

• Нам нужен способ отписаться от наблюдаемого источника. Я предполагаю, что когда результирующий наблюдатель будет завершен (или ошибочен), это подходящее время для отказа от подписки на источник, поскольку нашему наблюдателю было сказано ожидать больше событий.

Вот код:

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)); 
}