Introducing RX For Collections

Roland Pheasant
9y
9.7k
0
1

Article

Dynamic Data

Dynamic Data is a portable class library that brings the power of reactive (rx) to collections.

A collection that mutates can have adds, updates and removes. Out-of-the-box, rx does nothing to manage any changes in a collection. That is why Dynamic Data exists. In a Dynamic Data collection changes are notified via an observable change set that is the heart of the system. An operator receives these notifications and then applies some logic and subsequently provides it's own notifications. In this way operators can be chained together to apply powerful and often very complicated operations with some very simple fluent code.

The benefit of at least 40 operators, that are borne from pragmatic experience, is that the management of in-memory data becomes easy and it is no exaggeration to say it can save thousands of lines of code by abstracting complicated and often repetitive operations.

Getting started

As stated in the blurb, Dynamic Data is based on the concept of an observable change set. The easiest way to create one is directly from an observable.

IObservable<T> myObservable;
IObservable<IEnumerable<T>> myObservable;
//1. This option will create an observable where item's are identified using the hash code.
var mydynamicdatasource = myObservable.ToObservableChangeSet();
//2. Or specify a key like this
var mydynamicdatasource = myObservable.ToObservableChangeSet(t=> key);

The problem with the preceding is the collection will grow forever so there are overloads to specify size limitation or expiry times (not shown).

To have much more control over the root collection then we need a local data store that has the requisite crud methods. Like the preceding the cache can be created with or without specifying a key as in the following:

//1. Create a cache where item's are identified using the hash code.
var mycache = new SourceCache<TObject>();
//2. Or specify a key like this
var mycache = new SourceCache<TObject,TKey>(t => key);

One final out-of-the-box way to create an observable change set is if you are doing UI work and have an observable collection, you can do this:

var myobservablecollection= new ObservableCollection<T>();
//1. This option will create a collection where item's are identified using the hash code.
var mydynamicdatasource = myobservablecollection.ToObservableChangeSet();
//2. Or specify a key like this
var mydynamicdatasource = myobservablecollection.ToObservableChangeSet(t=> key);

One other point worth making here is any steam can be covered to as cache.

var mycache = somedynamicdatasource.AsObservableCache();

This cache has the same connection methods as a source cache but is read only.

Examples

Now you know how to create a Dynamic Data stream, here are a few quick fire examples based on the assumption that we already have an observable change set. In all of these examples the resulting sequences always exactly reflect the items is the cache, in other words adds, updates and removes are always propagated.

Example 1: filters a stream of live trades, creates a proxy for each trade and orders the result by the most recent first. As the source is modified the observable collection will automatically reflect changes.

//Dynamic data has it's own take on an observable collection (optimised for populating f
var list = new ObservableCollectionExtended<TradeProxy>();
var myoperation = somedynamicdatasource
.Filter(trade=>trade.Status == TradeStatus.Live)
.Transform(trade => new TradeProxy(trade))
.Sort(SortExpressionComparer<TradeProxy>.Descending(t => t.Timestamp))
.ObserveOnDispatcher()
.Bind(list)
.DisposeMany()
.Subscribe()

Also TradeProxy is disposable and DisposeMany() ensures items are disposed of when no longer part of the stream.

Example 2: produces a stream that is grouped by status. If an item's changes status it will be moved to the new group and when a group has no items the group will automatically be removed.

var myoperation = somedynamicdatasource
.Group(trade=>trade.Status) //This is NOT Rx's GroupBy
.Subscribe(changeSet=>//do something with the groups)

Example 3: Suppose I am editing some trades and I have an observable on each trades that validates but I want to know when all the items are valid then this will do the job.

IObservable<bool> allValid = somedynamicdatasource
.TrueForAll(trade => trade.IsValidObservable, (trade, isvalid) => isvalid)

This operator flattens the observables and returns the combined state in one line of code.

Example 4: will wire and un-wire items from the observable when they are added, updated or removed from the source.

var myoperation = somedynamicdatasource.Connect()
.MergeMany(trade=> trade.ObservePropertyChanged(t=>t.Amount))
.Subscribe(ObservableOfAmountChangedForAllItems=>//do something with IObservable<PropChangedArg>)

More Information

I could go on endlessly but this is not the place for full documentation. I promise this will come but for now I suggest downloading my WPF sample app (links above) since I intend it to be a "living document" and I promise it will be continually maintained.