如何实现从一个IObserver到另一个IObserver的原子切换？

Question

我有一个 IObservable，我使用一些中间步骤将其转换为 IObservable：

var observedXDocuments =
    from b in observedBytes
    // Lot of intermediate steps to transform byte arrays into XDocuments
    select xDoc;

在某个时间点，我对观察到 XDocument，因此我订阅了 IObserver。稍后，我想订阅另一个 IObserver并处理掉旧的。

如何在一个原子操作中完成此操作，而不丢失任何观察到的 XDocument？我可以这样做：

oldObserver.Dispose();
observedXDocuments.Subscribe(newObserver);

不过我担心，在这两个调用之间，我可能会丢失一个XDocument。如果我切换这两个调用，可能会两次收到相同的 XDocument。

Answer 1

我可能会添加一个间接层。编写一个名为 ExchangeableObserver 的类，将其订阅到您的可观察对象，并使其永久订阅。 ExchangeableObserver 的工作是将所有事情委托给给定的子观察者。但程序员可以随时更改委托的子观察者。在我的示例中，我有一个 Exchange() 方法。像这样的东西：

public class ExchangeableObserver<T> : IObserver<T> {
  private IObserver<T> inner;

  public ExchangeableObserver(IObserver<T> inner) {
    this.inner=inner;
  }

  public IObserver<T> Exchange(IObserver<T> newInner) {
    return Interlocked.Exchange(ref inner, newInner);
  }

  public void OnNext(T value) {
    inner.OnNext(value);
  }

  public void OnCompleted() {
    inner.OnCompleted();
  }

  public void OnError(Exception error) {
    inner.OnError(error);
  }
}

Answer 2

您可以使用一个信号量，确保在 IObservable 准备 IObservable 时不会发生观察者更改。

伪代码如何做到这一点（不是测试）

  System.Threading.ReaderWriterLockSlim criticalSection 
       = new System.Threading.ReaderWriterLockSlim(...);  


  ... converting from `IObservable<byte[]>` to `IObservable<XDocument>`  
  criticalSection.EnterReadLock();
  Call IObservable<XDocument>
  criticalSection.ExitReadLock();

  .... replacing IObservable<XDocument>
  criticalSection.EnterWriteLock();
  Call change IObservable<XDocument>
  criticalSection.ExitWriteLock();

编辑：使用Call IObservable

  > What exactly do you mean with the line `Call IObservable<XDocument>`?

我解释你的句子

  > I have an `IObservable<byte[]>` that I transform 
  > into an `IObservable<XDocument>` using some intermediate steps...

，你已经为IObservable注册了一个事件处理程序从 byte[] 创建一个 XDocument 然后调用
触发 IObservable事件的东西。

Call IObservable 表示触发后续事件的代码