如何在 ReactiveUI 中使用 WhenAny(...) 合并多个可观察对象？

Question

我有一个问题是本网站上提出的以下问题的延伸。

Is there a more elegant way to merge observables when return type is unimportant?

我有一个IObservable<Unit>（比如说X）、一个反应式集合（Y）和一个属性（Z）。返回类型并不重要。我只想在这些变化时订阅。

我知道如何使用Observable.Merge 观察所有 3 和 Subscribe，如下所示。

Observable.Merge(X, Y.Changed, ObservableForProperty(Z).Select(_ => Unit.Default)).Subscribe(..)

而且它有效。

但是，当我尝试使用WhenAny(...,....,....).Subscribe() 时，当我的X 更改时，订阅不会被触发。使用WhenAny(...) 而不是Observable.Merge(..) 执行上述操作的语法是什么？

我更喜欢使用WhenAny(....)，因为我在其他地方使用ReactiveUI。

示例： 假设我有一个派生自 ReactiveObject 的类，具有以下属性。

public class AnotherVM : ReactiveObject
{
    public bool IsTrue
    {
        get { return this.isTrue; }
        set { this.RaiseAndSetIfChanged(x => x.isTrue, ref this.isTrue, value); }
    }

    public IObservable<Unit> Data
    {
        get { return this.data; }
    }

    public ReactiveCollection MyCol
    {
       get { return Mycol; }
    }    
}

public class MyVM : ReactiveObject
{
    MyVM
    {
       // do WhenAny or Observable.Merge here....
    }
}

我想在AnotherVM 类中使用Observable.Merge(..) 或WhenAny(...) 在MyVM 类中观察上述属性。我发现当我使用WhenAny(...) 或Merge(...) 在MyVM 中订阅上述内容时，当三个属性中的任何一个发生更改时，我并不总是收到通知。

Answer 1

WhenAny 不是用于监控任意 observable 集合，而是用于监控 ReactiveUI 支持的对象的属性（如 ReactiveObject 或响应式集合）。

对于组合可观察流中的变化的一般情况，Observable.Merge 是正确的方法。

编辑

我注意到您已将 Data 和 MyCol 属性声明为只读。如果你像这样使用Merge：

Observerable.Merge(this.WhenAnyValue(o=>o.IsTrue, v=>Unit.Default),
                   this.Data,
                   this.MyCol.CollectionChanged.Select(v=>Unit.Default))

...那么您必须小心不要更改支持字段。如果这样做，那么您将丢失事件 - 也许这就是正在发生的事情？

在这种情况下，您需要将这些属性连接到 RaiseAndSetIfChanged 并使用 Switch 来跟踪 - 例如如果 this.data 可以更改，那么您将需要（我在这里使用 ReactiveUI 5 + .NET 4.5，以防 RaiseAndSetIfChanged 语法看起来很奇怪）：

public IObservable<Unit> Data
{
    get { return this.data; }
    private set { this.RaiseAndSetIfChanged(ref data, value); }
}

你的合并会是这样的：

Observerable.Merge(this.WhenAnyValue(o=>o.IsTrue, v=>Unit.Default),
                   this.WhenAnyObservable(x => x.Data),
                   this.MyCol.CollectionChanged.Select(v=>Unit.Default))

WhenAnyObservable 在概念上等价于：

WhenAny(x => x.Data, vm => vm.Value).Switch()

当它发生变化时，使用 Switch 翻转到 Data 的最新值。不要忘记使用 setter 来更改数据的值！

Answer 2

应该这样做。

IObservable<Unit> merged =
    Observerable.Merge
    ( this.WhenAnyValue(o=>o.IsTrue, v=>Unit.Default)
    , this.Data
    , this.MyCol.CollectionChanged.Select(v=>Unit.Default)
    )

理论上，您可以编写一个特殊版本的合并，它会忽略可观察对象的类型并返回IObservable<Unit>。然后你可以写

IObservable<Unit> merged =
    Observerable.MergeToUnit
    ( this.WhenAnyValue(o=>o.IsTrue)
    , this.Data
    , this.MyCol.CollectionChanged
    )

但是你需要许多 MergeToUnit 重载来支持最多 N 个参数。

WhenAny 与多个对象一起使用的最通用模式是

Observable.CombineLatest
  ( source0.WhenAnyValue(s=>s.FieldA)
  , source1.WhenAnyValue(s=>s.FieldB)
  , source2.WhenAnyValue(s=>s.FieldC)
  , source3.WhenAnyValue(s=>s.FieldD)
  , (a,b,c,d) => Process(a,b,c,d)
  )

有时习惯使用标准组合子会更好。