c# - 使用 IObservable (Rx) 作为 MVVM 的 INotifyCollectionChanged 替代品？

Question

我一直在研究在 MVVM 框架中使用 Rx。这个想法是对内存中的数据集使用“实时”LINQ 查询，将数据投影到 View 模型中进行绑定(bind)。

以前这可以通过使用 INotifyPropertyChanged/INotifyCollectionChanged 和一个名为 CLINQ 的开源库来实现. Rx 和 IObservable 的潜力是转移到更具声明性的 ViewModel，使用 Subject 类将更改的事件从源模型传播到 View。最后一步需要从 IObservable 转换为常规数据绑定(bind)接口(interface)。

问题是 Rx 似乎不支持实体已从流中删除的通知。示例如下。
该代码显示了一个 POCO，它使用 BehaviorSubject 类作为字段状态。代码继续创建这些实体的集合，并使用 Concat 将过滤器流合并在一起。这意味着对 POCO 的任何更改都会报告给单个流。

此流的过滤器设置为过滤 Rating==0。当偶数发生时，订阅只是将结果输出到调试窗口。

在任何元素上设置 Rating=0 将触发该事件。但是将 Rating 设置回 5 将看不到任何事件。

在 CLINQ 的情况下，查询的输出将支持 INotifyCollectionChanged - 因此从查询结果中添加和删除的项目将触发正确的事件以指示查询结果已更改(添加或删除的项目)。

我能想到的解决这个问题的唯一方法是设置两个具有相反(双重)查询的流。添加到相反流的项目意味着从结果集中删除。如果做不到这一点，我可以只使用 FromEvent 而不是使任何实体模型可观察——这使得 Rx 更像是一个事件聚合器。有什么指点吗？

using System;
using System.ComponentModel;
using System.Linq;
using System.Collections.Generic;

namespace RxTest
{

    public class TestEntity : Subject<TestEntity>, INotifyPropertyChanged
    {
        public IObservable<string> FileObservable { get; set; }
        public IObservable<int> RatingObservable { get; set; }

        public string File
        {
            get { return FileObservable.First(); }
            set { (FileObservable as IObserver<string>).OnNext(value); }
        }

        public int Rating
        {
            get { return RatingObservable.First(); }
            set { (RatingObservable as IObserver<int>).OnNext(value); }
        }

        public event PropertyChangedEventHandler PropertyChanged;

        public TestEntity()
        {
            this.FileObservable = new BehaviorSubject<string>(string.Empty);
            this.RatingObservable = new BehaviorSubject<int>(0);
            this.FileObservable.Subscribe(f => { OnNotifyPropertyChanged("File"); });
            this.RatingObservable.Subscribe(f => { OnNotifyPropertyChanged("Rating"); });
        }

        private void OnNotifyPropertyChanged(string property)
        {
            if (PropertyChanged != null) PropertyChanged(this, new PropertyChangedEventArgs(property));
            // update the class Observable
            OnNext(this);
        }

    }

    public class TestModel
    {
        private List<TestEntity> collection { get; set; }
        private IDisposable sub;

        public TestModel()
        {
            this.collection = new List<TestEntity>() {
            new TestEntity() { File = "MySong.mp3", Rating = 5 },
            new TestEntity() { File = "Heart.mp3", Rating = 5 },
            new TestEntity() { File = "KarmaPolice.mp3", Rating = 5 }};

            var observableCollection = Observable.Concat<TestEntity>(this.collection.Cast<IObservable<TestEntity>>());
            var filteredCollection = from entity in observableCollection
                                     where entity.Rating==0
                                     select entity;
            this.sub = filteredCollection.Subscribe(entity =>
                {
                    System.Diagnostics.Debug.WriteLine("Added :" + entity.File);
                }
            );
            this.collection[0].Rating = 0;
            this.collection[0].Rating = 5;
        }
    };
}

Answer 1

实际上，我发现 Reactive-UI 库对此很有帮助(在 NuGet 中可用)。 该库包括用于集合的特殊 IObservable 主题以及在传统 INCC 集合上创建这些“ReactiveCollections”之一的工具。 通过这个，我有新的、删除的项目和集合中更改项目的流。然后我使用 Zip 将流合并在一起并修改目标 ViewModel 可观察集合。这提供了基于对源模型的查询的实时投影。

下面的代码解决了这个问题(这段代码会更简单，但是 Reactive-UI 的 Silverlight 版本有一些问题需要解决方法)。该代码通过简单地调整其中一个集合元素的“评级”值来触发集合更改事件:

using System;
using System.ComponentModel;
using System.Linq;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Collections.Specialized;
using ReactiveUI;

namespace RxTest
{

    public class TestEntity :  ReactiveObject, INotifyPropertyChanged, INotifyPropertyChanging
    {
        public string _File;
        public int _Rating = 0;
        public string File
        {
            get { return _File; }
            set { this.RaiseAndSetIfChanged(x => x.File, value); }
        }

        public int Rating
        {
            get { return this._Rating; }
            set { this.RaiseAndSetIfChanged(x => x.Rating, value); }
        }

        public TestEntity()
        {
        }
    }

    public class TestModel
    {
        private IEnumerable<TestEntity> collection { get; set; }
        private IDisposable sub;

        public TestModel()
        {
            this.collection = new ObservableCollection<TestEntity>() {
            new TestEntity() { File = "MySong.mp3", Rating = 5 },
            new TestEntity() { File = "Heart.mp3", Rating = 5 },
            new TestEntity() { File = "KarmaPolice.mp3", Rating = 5 }};

            var filter = new Func<int, bool>( Rating => (Rating == 0));

            var target = new ObservableCollection<TestEntity>();
            target.CollectionChanged += new NotifyCollectionChangedEventHandler(target_CollectionChanged);
            var react = new ReactiveCollection<TestEntity>(this.collection);
            react.ChangeTrackingEnabled = true;

            // update the target projection collection if an item is added
            react.ItemsAdded.Subscribe( v => { if (filter.Invoke(v.Rating)) target.Add(v); } );
            // update the target projection collection if an item is removed (and it was in the target)
            react.ItemsRemoved.Subscribe(v => { if (filter.Invoke(v.Rating) && target.Contains(v)) target.Remove(v); });

            // track items changed in the collection.  Filter only if the property "Rating" changes
            var ratingChangingStream = react.ItemChanging.Where(i => i.PropertyName == "Rating").Select(i => new { Rating = i.Sender.Rating, Entity = i.Sender });
            var ratingChangedStream = react.ItemChanged.Where(i => i.PropertyName == "Rating").Select(i => new { Rating = i.Sender.Rating, Entity = i.Sender });
            // pair the two streams together for before and after the entity has changed.  Make changes to the target
            Observable.Zip(ratingChangingStream,ratingChangedStream, 
                (changingItem, changedItem) => new { ChangingRating=(int)changingItem.Rating, ChangedRating=(int)changedItem.Rating, Entity=changedItem.Entity})
                .Subscribe(v => { 
                    if (filter.Invoke(v.ChangingRating) && (!filter.Invoke(v.ChangedRating))) target.Remove(v.Entity);
                    if ((!filter.Invoke(v.ChangingRating)) && filter.Invoke(v.ChangedRating)) target.Add(v.Entity);
                });

            // should fire CollectionChanged Add in the target view model collection
            this.collection.ElementAt(0).Rating = 0;
            // should fire CollectionChanged Remove in the target view model collection
            this.collection.ElementAt(0).Rating = 5;
        }

        void target_CollectionChanged(object sender, NotifyCollectionChangedEventArgs e)
        {
            System.Diagnostics.Debug.WriteLine(e.Action);
        }
    }
}