使用 IObservable<T> 跟踪当前状态

Question

假设我有一个观察 IObservable 的对象，以便它始终知道某个外部源的当前状态。在内部，我的对象有一个方法使用该外部值作为操作的一部分：

public class MyObject
{
  public MyObject(IObservable<T> externalSource) { ... }

  public void DoSomething()
  {
    DoSomethingWith(CurrentT);
  }
}

使用 IObservable 来“跟踪当前状态”而不是“响应事件流”的惯用“反应式”方式是什么。

想法 #1 是只监视可观察的值并在它们进入时记下它们。

public class MyObject
{
  private T CurrentT;
  public MyObject(IObservable<T> externalSource) 
  {
    externalSource.Subscribe((t) => { CurrentT = t; });
  }

  public void DoSomething()
  {
    DoSomethingWith(CurrentT);
  }
}

这很好，但是跟踪类成员中的状态似乎非常不具反应性。

想法 #2 是使用BehaviorSubject

public class MyObject
{
  private readonly BehaviorSubject<T> bs;
  public MyObject(BehvaiorSubject<T> externalSource) 
  {
    this.bs = externalSource
  }

  public void DoSomething()
  {
    DoSomethingWith(bs.Value);
  }
}

但是直接使用主题似乎是不受欢迎的。但至少在这种情况下，我有能力使用只读字段来存储行为主体。

BehaviorSubject（或 ReplaySubject）看起来确实是为此目的而制作的，但这里还有其他更好的方法吗？如果我应该使用主题，将主题作为注入参数，还是将原始可观察对象并在构造函数中本地构建主题更有意义？

（顺便说一句，如果源 observable 尚未触发，我知道需要处理第一个值。不要挂断电话，这不是我要问的）

Answer 1

要成为 Rx-ish，我建议避免使用第二个选项并使用第一个选项，但可以通过两种方式之一进行修改。

要么 (1) 让你的类成为一次性的，这样你就可以干净地关闭对 observables 的订阅，或者 (2) 制作一个让你清理单个 observables 的方法。

(1)

public class MyObject : IDisposable
{
    private T CurrentT;
    private IDisposable Subscription;
    public MyObject(IObservable<T> externalSource) 
    {
        Subscription = externalSource
            .Subscribe((t) => { CurrentT = t; });
    }

    public void Dispose()
    {
        Subscription.Dispose();
    }

    public void DoSomething()
    {
        DoSomethingWith(CurrentT);
    }
}

(2)

public class MyObject
{
    private T CurrentT;

    public IDisposable Observe(IObservable<T> externalSource) 
    {
        return externalSource
            .Subscribe((t) => { CurrentT = t; });
    }

    public void DoSomething()
    {
        DoSomethingWith(CurrentT);
    }
}

两者都允许适当的清理，并且都不使用主题。

Answer 2

我会选择使用ReactiveUI library 的通用解决方案。 RUI 有一种将IObservable<T> 映射到INotifyPropertyChanged 有状态属性的标准方法。

public class ObservableToINPCObject<T> : ReactiveObject, IDisposable
{
    ObservableAsPropertyHelper<T> _ValueHelper;
    public T Value {
        get { return _ValueHelper.Value; }
    }

    public ObservableToINPCObject(IObservable<T> source, T initial = default(T))
    {
        _ValueHelper = source.ToProperty(this, p=>p.Value, initial);
    }

    public Dispose(){
        _ValueHelper.Dispose();
    }
}

ValueHelper 既包含 observable 的 current state，又在状态改变时自动触发正确的 INPC 通知。这是为你处理的相当多的样板。

还有一个扩展方法

public static class ObservableToINPCObject {
    public static ObservableToINPCObject<T> ToINPC<T>
        ( this IObservable<T> source, T init = default(T) )
        {
            return new ObservableToINPCObject(source, init);
        }
}

现在给出一个

IObservable<int> observable;

你可以的

var obj = observable.ToINPC(10);

获取最新值

Console.WriteLine(obj.Value);

还考虑到 Value 是 INPC 支持属性，您可以在数据绑定中使用它。我一直使用 ToProperty 将我的 observables 公开为 WPF 数据绑定的属性。