c# - 使用 Rx 构建可观察的存储库

Question

我在常见的场景中工作，我想访问存储库的一个子集而不用担心必须保持更新，例如'获取所有价格大于10的订单'。我已经实现了一个解决方案，但有两个问题(列在最后)。

存储库的一个子集可以用等同于

的东西来实现

var expensiveOrders = Repository.GetOrders().Where(o => o.Price > 10);

但这是一个IEnumerable，不会在原始集合更新时更新。我可以为 CollectionChanged 添加处理程序，但如果我们想访问更多的子集怎么办？

var expensiveOrdersFromBob = expensiveOrders.Where(o => o.Name == Bob);

我们还必须为此连接一个集合更改。实时更新的概念让我想到了 Rx，因此我着手构建一个 ObservableCache，其中包含自动更新自身的项目的 ObservableCollection 和一个 RX流通知。 (流也是更新引擎盖下缓存的内容。)

class ObservableCache<T> : IObservableCache<T>
{
    private readonly ObservableCollection<T> _cache;
    private readonly IObservable<Tuple<T, CRUDOperationType>> _updates;

    public ObservableCache(IEnumerable<T> initialCache
       , IObservable<Tuple<T, CRUDOperationType>> currentStream, Func<T, bool> filter)
        {
          _cache = new ObservableCollection<T>(initialCache.Where(filter));
          _updates = currentStream.Where(tuple => filter(tuple.Item1));
          _updates.Subscribe(ProcessUpdate);
        }

    private void ProcessUpdate(Tuple<T, CRUDOperationType> update)
    {
        var item = update.Item1;
        lock (_cache)
        {
            switch (update.Item2)
            {
                case CRUDOperationType.Create:
                    _cache.Add(item);
                    break;
                case CRUDOperationType.Delete:
                    _cache.Remove(item);
                    break;
                case CRUDOperationType.Replace:
                case CRUDOperationType.Update:
                    _cache.Remove(item); // ToDo: implement some key-based equality
                    _cache.Add(item);
                    break;
            }
        }
    }

    public ObservableCollection<T> Cache
    {
        get { return _cache; }
    }

    public IObservable<T> Updates
    {
        get { return _updates.Select(tuple => tuple.Item1); }
    }

    public IObservableCache<T> Where(Func<T, bool> predicate)
    {
        return new ObservableCache<T>(_cache, _updates, predicate);
    }
}

然后你可以像这样使用它:

var expensiveOrders = new ObservableCache<Order>(_orders
                                                 , updateStream
                                                 , o => o.Price > 10);
expensiveOrders.Updates.Subscribe
     (o => Console.WriteLine("Got new expensive order: " + o));
_observableBoundToSomeCtrl = expensiveOrders.Cache;

var expensiveOrdersFromBob = expensiveOrders
                             .Where(o => o.Name == "Bob");
expensiveOrdersFromBob.Updates.Subscribe
         (o => Console.WriteLine("Got new expensive order from Bob: " + o));
_observableBoundToSomeOtherCtrl = expensiveOrdersFromBob.Cache;

等等，这个想法是您可以将缓存投影到越来越窄的子集中，而不必担心它会不同步。那我的问题是什么？

1. 我想知道我是否可以通过让 RX 本质上更新集合来消除 CRUD 问题。也许用 Select 或类似的东西“转换”更新？
2. 存储库与更新模式固有的竞争条件，因为我在构建新缓存时可能会错过一些更新。我想我需要某种排序，但这意味着让我的所有 T 对象都实现一个 ISequenceableItem 接口(interface)。有没有更好的方法来做到这一点？ RX 很棒，因为它为您处理所有线程。我想利用这一点。

Answer 1

位于 http://github.com/wasabii/OLinq 的 OLinq 项目专为这种响应式(Reactive)更新而设计，我认为 ObservableView 就是您所追求的。