c# – 如何实现从一个IObserver到另一个IObserver的原子开关？

发布时间：2020-12-16 01:44:44 所属栏目：百科来源：网络整理

导读：我有一个IObservable byte []我转换成IObservable XDocument使用一些中间步骤： var observedXDocuments = from b in observedBytes // Lot of intermediate steps to transform byte arrays into XDocuments select xDoc; 在某个时间点,我对观察到的XDocume

我有一个IObservable< byte []>我转换成IObservable< XDocument>使用一些中间步骤：

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

在某个时间点,我对观察到的XDocuments感兴趣所以我订阅了IObserver< XDocument>.在稍后的时间点,我想订阅另一个IObserver< XDocument>并处理旧的.

如何在一次原子操作中执行此操作,而不会丢失任何观察到的XDocument？我可以这样做：

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

我很担心,在这两个电话之间,我可以放松一个XDocument.如果我切换两个调用,可能会发生两次我收到相同的XDocument.

解决方法

我可能会添加一层间接.编写一个名为ExchangeableObserver的类,将其订阅到您的observable,并保持永久订阅. 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);
  }
}

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!