DEV Community: Nick Polyak

Introduction to Dynamic Data's Observable Cache

Nick Polyak — Sun, 04 May 2025 17:37:21 +0000

Intro to Dynamic Data

Dynamic Data (DD) applies the concepts of Reactive Observable (Rx) Streams to collections.

DD was created by Roland Pheasant for the purpose of greatly simplifying the data collection updates.

It allows monitoring the source collections and updating target collection(s) in a way that the target collection satisfies conditions specified by the Dynamic Data operators.

For example assuming that we have a source collection of Trade objects representing pending trades for various stocks. We can filter the target collection to contain only Trades for IBM stock.

Filtering is only one Dynamic Data operator out of many. Same as LINQ or Rx operators, DD operators can be piped together making them perform as complex tasks as needed.

Essentially all LINQ operators have their matches in Dynamic Data (in Dynamic Data operators have different names in order to avoid clashes with Rx operators). For example

Rx (or LINQ) Operator Where(...) is named Filter(...) in DD.
Select(...) operator is named Transform(...) in DD.
OrderBy(...) is called Sort(...) in DD.
GroupBy(...) is named Group(...) in DD.

Dynamic Data has two special types representing the observable collections that can serve as source collections:

SourceList<T> for collections whose items are immutable (or at least whose items changes do not need to be observed by the DD operators).
SourceCache<T, TKey> - for collections with updatable items.

In this article we shall talk about SourceCache<T, TKey> containing updatable objects (while the previous article in the series was dedicated to SourceLists.

Source Code Location

All the source code for this article is located under Intro to Observable Cache within NP.Sample repository.

The samples are built as XUnit Facts defined within
SimpleObservableCacheExamples class. Each of the facts can be run separately from the rest.

For more on XUnit read Everything you need to know to create XUnit C# Tests in Visual Studio.

Simplified `Trade` class

All our samples are built around Trade class representing a simplification of a stock trade:

public class Trade : 
    AbstractNotifyPropertyChanged
{
    // primary key (used for distinguishing 
    // between the new entries and the 
    // updates_)
    public int TradeId { get; }

    // Stock Symbol
    public Symbol TheSymbol { get; }

    // updatable total trade prices
    decimal _totalTradePrice;
    public decimal TotalTradePrice 
    {
        get => _totalTradePrice; 

        // SetAndRaise fires PropertyChanged event
        // when TotalTradePrice property changes
        set => SetAndRaise(ref _totalTradePrice, value);
    }

    public Trade
    (
        int tradeId,
        Symbol symbol,
        decimal totalTradeAmount)
    {
        this.TradeId = tradeId;
        this.TheSymbol = symbol;
        this.TotalTradePrice = totalTradeAmount;
    }

    ...

    public override bool Equals(object? obj)
    {
        if (obj is Trade trade)
        {
            return this.TradeId.Equals(trade.TradeId) && 
                   (this.TheSymbol == trade.TheSymbol) && 
                   (this.TotalTradePrice == trade.TotalTradePrice);
        }

        return false;
    }
}

Trade has only 3 properties:

int TradeId { get; }
Symbol TheSymbol { get; }
decimal TotalTradePrice { get { ... } set {...} }

The first two properties TradeId and TheSymbol cannot be modified, while the 3rd TotalTradePrice can be updated and fires INotifyPropertyChanged.PropertyChanged event after it is updated.

Property TradeId is the primary key - it determines if the trade is a new trade within the cache or an update of an existing trade. If a Trade with the same TradeId already exists in the SourceCache<Trade, int> then the newly coming entry is an update, otherwise it is a new Trade object.

Property TheSymbol uniquely specifies the stock of the trade. Note, that there can be different trades with for the same stock at the same time within the system. This property is of type Symbol which is an C# enum (since this sample deals with very few stocks, I wanted to take advantage of enum's strong typing instead of using strings which leave a strong changes of spelling errors).

Property TotalTradePrice essentially represents the number of shares multiplied by the per share price for the trade.

Those who deal with stock trades will easily understand that this Trade class is greatly simplified. I already mentioned number-of-shares and price-per-share properties are missing. We also do not specify whether it is a Buy or Sell trade (let's assume they are all Sells), trade timing, trade status and other possible data.

Basic `SourceCache<T, TKey>` API

Here we shall talk about most important SourceCache<T, TKey> API needed used within our samples below.

As was mentioned above, all the samples are build around the Trade class with int TradeId property playing the role of the primary key for specifying whether an update or a new trade is coming.

Creating `SourceCache<Trade, int>`

To create the SourceCache<Trade, int> use the following constructor:

var tradeSourceCache = new SourceCache<Trade, int>(t => t.TradeId);

The argument to the constructor (in our case it is t => t.TradeId) is a lamba expression of type Func<Trade, int> specifying how to obtain the unique trade key from a Trade object (remember we use this key to figure out if we deal with an update or with the new trade).

Adding or Updating a Trade to Cache

To add or update a trade (depending on whether the trade with the same TradeId already exists in the cache), we use the method sourceCache.AddOrUpdate(Trade trade), e.g.:

Trade metaTrade = 
      new Trade 
     { 
         TradeId = 21, 
         Trade = Symbol.META,
         TotalTradePrice = 2000
     };

sourceCache.AddOrUpdate(metaTrade);

The sourceCache itself will check a Trade with TradeId equals 21 already exists in it. If yes, it will do an update, if no, it will insert a new Trade object.

Using `IObservableCache<T, TKey>`

IObservableCache<T, TKey> has all the functionality for observing changes to the SourceCache, but does not have any functionality for modifying it. In a sense it is a read-only view into SourceCache<T, TKey>.

You might want to pass IObservableCache<T, TKey> to methods that do not modify the cache. To convert SourceCache into IObservableCache, the best is to use AsObservableCache() method:

IObservableCache<Trade, int> observableCache = 
                                 sourceCache.AsObsrevableCache();

Turning `SourceCache<T, TKey>` (or `IObservableView<T, TKey>`) into an Rx Observable Stream of Changes to the Collection using `Connect()` Method

Same as ObservableList<T> or SourceList<T> discussed in the previous installment of the series, all the composable LINQ-like operators should be applied not to the SourceCache<T, TKey> but to the Rx Stream of collection changes IObservable<IChangeSet<T, TKey>>.

To convert SourceCache (or ObservableCache) into the Rx Stream of collection changes, use Connect() method, e.g.:

IObservable<IChangeSet<Trade, int>> changeSetStream =
                                      sourceCache.Connect();

Turning `ObservableCollection<T>` into an Rx Observable Stream of Collection Changes

ObservableCollection<T> is a special collection that implements INotifyCollectionChanged interface. After every change to the collection it fires INotifyCollectionChanged.CollectionChanged event passing the information about the change to the event's arguments.

People who work with the MVVM pattern and with WPF, Avalonia or other XAML framework, know how useful ObservableCollections are - they provide a collection within a view models whose visuals will be changed automatically as elements are added, removed or updated in it.

Dynamic Data provides method ToObservableChangeSet(...) allowing to convert an ObservableCollection<T> straight into the Rx Observable Stream of Collection Changes without using SourceCache explicitly, e.g.:

ObservableCollection<Trade> myObservableCollection = 
                              new ObservableCollection<Trade>();
IObservable<IChangeSet<Trade, int>> changeSetStream =
          myObservableCollection.ToObservableChangeSet(t => t.TradeId);

Once the stream is subscribed to, whatever changes are made to myObservableCollection will be reflected within changeObservableStream.

Code Samples

Now we are ready for the code samples.

As was mentioned above, the code sample solution is IntoToObservableCache.sln from Intro to Observable Cache folder within NP.Samples repository.

All the sample are created as XUnit static [Fact] methods within SimpleObservableCacheExamples.cs file.

Auxiliary Methods and APIs used throughout the Tests

As was mentioned above, all the stock symbols were defined as enum Symbol for the sake of strong typing :

public enum Symbol
{
    META,
    ORCL,
    TSLA,
    MSFT
}

Static class TradesGenerator contains an extension method Create(...) which returns a new trade or a trade update:

internal static Trade CreateTrade
(
    this Symbol symbol,
    decimal totalTradeAmount,
    int tradeId = -1
)
{
    ...
}

If the passed argument tradedId is default (-1), then a new trade with new TradeId is generated. Otherwise the user of the method can pass whatever non-negative value for tradeId that he wants and if such value already exists in the SourceCache - this trade will be considered an update.

SimpleObservableCacheExamples static class contains a static method Verify which compares two collections:

private static void Verify
(
    this IEnumerable<Trade> sourceTradeCollection,
    IEnumerable<Trade> targetCollection, 
    Func<Trade, bool> sourceFilter
)
{
    var filteredAndOrdered =
        sourceTradeCollection
        .Where(sourceFilter)
        .OrderBy(t => t.TotalTradePrice)
        .ToList();

    Assert.True(filteredAndOrdered.SequenceEqual(targetCollection));
}

One of these collection (sourceTradeCollection) is the source collection of trades, the other collection (targetCollection) is the result of applying Dynamic Data operators.

The purpose of this method is to assert that both collections have the same data arranged in the same order.

We assume that our Dynamic Data operators are always sorting the targetCollection in ascension order.

Also we pass a filter to be applied to the source collection (should be the same filter as we apply to obtain the targetCollection).

Filtering, Sorting and Binding Test

Here is the code for FilterSortAndBindTest() method (do not try to understand it all at once, since we shall describe it piece by piece below).

[Fact]
public static void FilterSortAndBindTest()
{
    // create sourceCache
    ISourceCache<Trade, int>? sourceCache =
        new SourceCache<Trade, int>(t => t.TradeId);

    // create several trades and add them to 
    // the sourceCache
    var metaTrade1 = Symbol.META.CreateTrade(2000);
    sourceCache.AddOrUpdate(metaTrade1);

    var oracleTrade1 = Symbol.ORCL.CreateTrade(1000);
    sourceCache.AddOrUpdate(oracleTrade1);

    var metaTrade2 = Symbol.META.CreateTrade(1900);
    sourceCache.AddOrUpdate(metaTrade2);

    var oracleTrade2 = Symbol.ORCL.CreateTrade(900);
    sourceCache.AddOrUpdate(oracleTrade2);

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<Trade, int>> changeSetStream =
        sourceCache!.Connect();

    // create the target collection
    IObservableCollection<Trade> targetCollection =
        new ObservableCollectionExtended<Trade>();

    // filter, sort and bind the source collection
    // to the targetCollection
    IObservable<IChangeSet<Trade, int>> resultObservable =
        changeSetStream
            .Filter(t => t.TheSymbol == Symbol.ORCL)
            .SortAndBind
            (
                targetCollection,
                SortExpressionComparer<Trade>
                     .Ascending(t => t.TotalTradePrice)
            );

    // now subscribe to start pulling data
    // using clause will dispose the subscription
    using var subscription = resultObservable.Subscribe();

    // create the original source collection
    IEnumerable<Trade> sourceTradeCollection =
        [metaTrade1, oracleTrade1, metaTrade2, oracleTrade2];

    // assert that the filter, and sort operators
    // result in the correct targetCollection
    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    // create another entry with the same TradeId key
    // as oracleTrade1 and add it to the sourceCache
    // to override the previous oracleTrade1 entry
    var oracleTrade1_modified =
        Symbol.ORCL.CreateTrade(100, oracleTrade1.TradeId);
    sourceCache.AddOrUpdate(oracleTrade1_modified);

    // change the sourceTradeCollection to contain
    // oracleTrade1_modified trade instead of oracleTrade1
    sourceTradeCollection =
        [metaTrade1, oracleTrade1_modified, metaTrade2, oracleTrade2];

    // make sure the collections are identical
    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    // create a new trade oracleTrade3 and add it 
    // to the sourceCache
    var oracleTrade3 =
        Symbol.ORCL.CreateTrade(50);
    sourceCache.AddOrUpdate(oracleTrade3);

    // create the new sourceTradeCollection (adding
    // oracleTrade3
    sourceTradeCollection =
        [
            metaTrade1,
            oracleTrade1_modified,
            metaTrade2,
            oracleTrade2,
            oracleTrade3
        ];

    // verify that the targetCollection is correct
    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    // remove all oracle entries
    sourceCache.Remove(oracleTrade1.TradeId);
    sourceCache.Remove(oracleTrade2.TradeId);
    sourceCache.Remove(oracleTrade3.TradeId);

    // make sure the targetCollection is empty
    Assert.True(targetCollection.Count() == 0);
}

First, we create the source cache

ISourceCache<Trade, int> sourceCache =
            new SourceCache<Trade, int>(t => t.TradeId);

Then we create 4 trades (all of them are new trades, none are updates) and add each one of them to the cache:

// create a meta trade with TotalTradePrice 2000
var metaTrade1 = Symbol.META.CreateTrade(2000);
// add metaTrade1 to the sourceCache
sourceCache.AddOrUpdate(metaTrade1);

var oracleTrade1 = Symbol.ORCL.CreateTrade(1000);
sourceCache.AddOrUpdate(oracleTrade1);

var metaTrade2 = Symbol.META.CreateTrade(1900);
sourceCache.AddOrUpdate(metaTrade2);

var oracleTrade2 = Symbol.ORCL.CreateTrade(900);
sourceCache.AddOrUpdate(oracleTrade2);

Then we create the Rx Observable Stream of cache changes:

IObservable<IChangeSet<Trade, int>> changeSetStream =
    sourceCache.Connect();

Now, we create the target collection that will contain the DD result:

IObservableCollection<Trade> targetCollection =
    new ObservableCollectionExtended<Trade>();

IObservableCollection<T> and ObservableCollectionExtended<T> are both introduced by DD and represent the same functionality as ObservableCollection<T> with some extra capability (e.g. batching multiple updates).

Now we apply the filter, sort and bind operations (bind will place the output into the targetCollection):

IObservable<IChangeSet<Trade, int>> resultObservable =
    changeSetStream
        .Filter(t => t.TheSymbol == Symbol.ORCL)
        .SortAndBind
        (
            targetCollection,
            SortExpressionComparer<Trade>
                      .Ascending(t => t.TotalTradePrice)
        );

We filter only trades with ORCL symbol and sort them by TotalTradePrice in ascending order.

Note, to improve the performance we use method SortAndBind(...) instead of two separate DD methods Sort(...) and Bind(...).

Now, to start pulling the data, we call Subscribe() method:

using IDisposable subscription = resultObservable.Subscribe();

using clause ensures that the subscription is destroyed automatically when the test goes out of scope.

We build the sourceTradeCollection and verify that under trade.TheSymbol == Symbol.ORCL filter, the collections match:

IEnumerable<Trade> sourceTradeCollection =
    [metaTrade1, oracleTrade1, metaTrade2, oracleTrade2];
sourceTradeCollection
    .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

Next we test that an update to the first oracle trade works - we create a trade update with the same TradeId and TheSymbol as oracleTrade1 and with TotalTradePrice equal 100, then we use it to update the sourceCache and we check that this update is reflected in the targetCollection:

var oracleTrade1_modified =
    Symbol.ORCL.CreateTrade(100, oracleTrade1.TradeId);
sourceCache.AddOrUpdate(oracleTrade1_modified);

sourceTradeCollection =
    [metaTrade1, oracleTrade1_modified, metaTrade2, oracleTrade2];
sourceTradeCollection
    .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

Next we check that once we add a trade to the source, it is also added to the target (if it is filtered in):

 // new trade
 var oracleTrade3 =
     Symbol.ORCL.CreateTrade(50);
 sourceCache.AddOrUpdate(oracleTrade3);

 sourceTradeCollection =
     [
         metaTrade1,
         oracleTrade1_modified,
         metaTrade2,
         oracleTrade2,
         oracleTrade3
     ];

 // verify that the targetCollection is correct
 sourceTradeCollection
     .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

Now, we shall remove all the oracle entries from the sourceCache and assert that the target collection is empty (since targetCollection contains only oracle entries because of the filter):

// remove all oracle entries
sourceCache.Remove(oracleTrade1.TradeId);
sourceCache.Remove(oracleTrade2.TradeId);
sourceCache.Remove(oracleTrade3.TradeId);

// make sure the targetCollection is empty
Assert.True(targetCollection.Count() == 0);

Filtering, Sorting and Binding Test with `ObservableCollection<Trade>` as Source.

Our next test defined by the static method FilterSortAndBindFromObservableCollectionTest() is almost exactly the same as the previous one (defined by FilterSortAndBindTest()) only instead of SourceCache<Trade, int>, we are using an ObservableCollection<Trade> as the source:

[Fact]
public static void 
    FilterSortAndBindFromObservableCollectionTest()
{
    // create individual trades
    var metaTrade1 = Symbol.META.CreateTrade(2000);

    var oracleTrade1 = Symbol.ORCL.CreateTrade(1000);

    var metaTrade2 = Symbol.META.CreateTrade(1900);;

    var oracleTrade2 = Symbol.ORCL.CreateTrade(900);

    // create the input ObservableCollection<Trade>
    ObservableCollection<Trade> sourceTradeCollection =
        new ObservableCollection<Trade>
        {
            metaTrade1, oracleTrade1, metaTrade2, oracleTrade2
        };

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<Trade, int>> changeSetStream =
        sourceTradeCollection.ToObservableChangeSet(t => t.TradeId);

    // create the target collection
    IObservableCollection<Trade> targetCollection =
        new ObservableCollectionExtended<Trade>();

    IObservable<IChangeSet<Trade, int>> resultObservable =
        changeSetStream
            .Filter(t => t.TheSymbol == Symbol.ORCL)
            .AutoRefresh() // listens to PropertyChanged
                            // events and updates 
                            // downstream results when 
                            // PropertyChanged is fired
            .SortAndBind
            (
                targetCollection,
                SortExpressionComparer<Trade>
                        .Ascending(t => t.TotalTradePrice)
            );

    // now subscribe to start pulling data
    // using clause will dispose the subscription
    using var subscription = resultObservable.Subscribe();

    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    // update the source entry and 
    // make sure that the target got resourced
    oracleTrade1.TotalTradePrice = 100;

    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    // add another oracle trade
    var oracleTrade3 =
        Symbol.ORCL.CreateTrade(50);

    sourceTradeCollection.Add(oracleTrade3);

    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    // remove all oracle entries
    sourceTradeCollection.Remove(oracleTrade1);
    sourceTradeCollection.Remove(oracleTrade2);
    sourceTradeCollection.Remove(oracleTrade3);

    // the target collection should become empty
    Assert.True(targetCollection.Count == 0);
}

Below, we shall go over only differences between this and the previous example.

The first difference is that instead of creating a SourceCache<Trade, int>, we create a ObservableCollection<Trade> and populate it with trades:

// create individual trades
var metaTrade1 = Symbol.META.CreateTrade(2000);

var oracleTrade1 = Symbol.ORCL.CreateTrade(1000);

var metaTrade2 = Symbol.META.CreateTrade(1900);;

var oracleTrade2 = Symbol.ORCL.CreateTrade(900);

// create the input ObservableCollection<Trade>
ObservableCollection<Trade> sourceTradeCollection =
    new ObservableCollection<Trade>
    {
        metaTrade1, oracleTrade1, metaTrade2, oracleTrade2
    };

Next we use ToObservableChangeSet(...) extension method to create the Rx observable stream of collection changes:

Observable<IChangeSet<Trade, int>> changeSetStream =
   sourceTradeCollection.ToObservableChangeSet(t => t.TradeId);

Among our DD operators we insert AutoRefresh() which will update the sorting every time PropertyChanged event is fired on any on the collection trade objects (which would happen when its TotalTradePrice changes):

IObservable<IChangeSet<Trade, int>> resultObservable =
    changeSetStream
        .Filter(t => t.TheSymbol == Symbol.ORCL)
        .AutoRefresh() // listens to PropertyChanged
                       // events and updates 
                       // downstream results when 
                       // PropertyChanged is fired
        .SortAndBind
        (
            targetCollection,
            SortExpressionComparer<Trade>
                            .Ascending(t => t.TotalTradePrice)
        );

To update a trade, instead of adding another Trade object with the same TradeId to the cache, update the instance within the ObservableCollection<Trade>:

oracleTrade1.TotalTradePrice = 100;

This will trigger the PropertyChanged event on the object, which will be handled by the AutoRefresh() method and passed to the downstream sorting operator.

Dynamic Filter Test

This test will demonstrate using dynamic filter that can be easily changed at run time.

In this test we shall use the ObservableCollection<Trade> source instead of the SourceCache<Trade, int>.

Here is the full code for the [Fact] public static void DynamicFilterTest() test method:

[Fact]
public static void
    DynamicFilterTest()
{
    // create trades
    var metaTrade1 = Symbol.META.CreateTrade(2000);

    var oracleTrade1 = Symbol.ORCL.CreateTrade(1000);

    var metaTrade2 = Symbol.META.CreateTrade(1900); ;

    var oracleTrade2 = Symbol.ORCL.CreateTrade(900);

    // creates the source ObservableCollection<Trade>
    ObservableCollection<Trade> sourceTradeCollection =
        new ObservableCollection<Trade>
        {
            metaTrade1, oracleTrade1, metaTrade2, oracleTrade2
        };

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<Trade, int>> changeSetStream =
        sourceTradeCollection.ToObservableChangeSet(t => t.TradeId);

    // create filterObservable subject to allow
    // changing the filters
    Subject<Func<Trade, bool>> filterObservable = 
        new Subject<Func<Trade, bool>>();

    // create the target collection
    IObservableCollection<Trade> targetCollection =
        new ObservableCollectionExtended<Trade>();

    IObservable<IChangeSet<Trade, int>> resultObservable =
        changeSetStream
            // use the Filter(...) method
            // that accepts filterObservable
            .Filter(filterObservable)
            .AutoRefresh()
            .SortAndBind
            (
                targetCollection,
                SortExpressionComparer<Trade>
                            .Ascending(t => t.TotalTradePrice)
            );

    // now subscribe to start pulling data
    // using clause will dispose the subscription
    using var subscription = resultObservable.Subscribe();

    // push the lambda expression to filter in only
    // ORCL symbol into filterObservable
    filterObservable.OnNext(t => t.TheSymbol == Symbol.ORCL);

    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    oracleTrade1.TotalTradePrice = 100;

    sourceTradeCollection
        .Verify(targetCollection, t => t.TheSymbol == Symbol.ORCL);

    // clear the source collection of oracle 
    // entries
    sourceTradeCollection.Remove(oracleTrade1);
    sourceTradeCollection.Remove(oracleTrade2);

    Assert.True(targetCollection.Count == 0);

    // change filter to META
    filterObservable.OnNext(t => t.TheSymbol == Symbol.META);

    // we should get some META entries in the target Collection
    // right away
    Assert.Contains<Trade>
        (targetCollection, t => t.TheSymbol == Symbol.META);
}

Now we shall go over the main differences between this and the previous samples.

The major difference is that we create an Rx subject to push various filters of type Func<Trade, bool> into it:

Subject<Func<Trade, bool>> filterObservable = 
    new Subject<Func<Trade, bool>>();

Then we use the Filter(...) extension method that accepts IObservable<Func<T, bool>> as the filtering operator:

IObservable<IChangeSet<Trade, int>> resultObservable =
    changeSetStream
        // use the Filter(...) method
        // that accepts filterObservable
        .Filter(filterObservable)
        .AutoRefresh()
        .SortAndBind
        (
            targetCollection,
            SortExpressionComparer<Trade>
                      .Ascending(t => t.TotalTradePrice)
        );

Note - .Filter(filterObservable) line.

Then, after we start pulling data from the changeSetStream by calling using var subscription = resultObservable.Subscribe();, we push the first filter we want into the filterObservable subject to filter only ORCL symbols:

filterObservable.OnNext(t => t.TheSymbol == Symbol.ORCL);

At the end of the test method, when we want to change the filter to say filtering in META symbols, we call:

filterObservable.OnNext(t => t.TheSymbol == Symbol.META);

and then assert that there are some trades with META symbol in the output collection:

// we should get some META entries in the target Collection
// right away
Assert.Contains<Trade>
    (targetCollection, t => t.TheSymbol == Symbol.META);

Grouping Test

Grouping test shows how to use Dynamic Data operators for grouping trades by Symbol.

Static test method for grouping is called GroupingTest():

[Fact]
public static void GroupingTest()
{
    // create trades
    var metaTrade1 = Symbol.META.CreateTrade(2000);

    var oracleTrade1 = Symbol.ORCL.CreateTrade(1000);

    var metaTrade2 = Symbol.META.CreateTrade(1900); ;

    var oracleTrade2 = Symbol.ORCL.CreateTrade(900);

    // create the source collection of trades as
    // observable collection
    ObservableCollection<Trade> sourceTradeCollection =
        new ObservableCollection<Trade>
        {
            metaTrade1, oracleTrade1, metaTrade2, oracleTrade2
        };

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<Trade, int>> changeSetStream = 
        sourceTradeCollection.ToObservableChangeSet(t => t.TradeId);

    // do the grouping
    IObservable<IGroupChangeSet<Trade, int, Symbol>>
        groupedObservable =
            changeSetStream
                .AutoRefresh()
                .Group(t => t.TheSymbol);

    // transform the grouped entries into 
    // SymbolTradeGroup objects
    var transformedGroups =
        groupedObservable.Transform(g => new SymbolTradeGroup(g));

    //output groups
    ReadOnlyObservableCollection<SymbolTradeGroup>? symbolTradeGroups;

    // create and populate an observable collection
    // symbolTradeGroups that contains those SymbolTradeGroup
    // objects
    using IDisposable subscription =
        transformedGroups
            .Bind(out symbolTradeGroups) // create and populate
            .DisposeMany() // make sure that if an item is removed
                            // from the collection, it is disposed
            .Subscribe();  // start the subscription

    // Assert there are two groups
    // (one for ORCL and the other for META trades)
    Assert.True(symbolTradeGroups.Count == 2);

    // get the oracleGroup and assert it exists and single
    SymbolTradeGroup oracleGroup = 
        symbolTradeGroups.Single(g => g.TheSymbol == Symbol.ORCL);

    Assert.True(oracleGroup.Trades.Count() == 2);
    Assert
        .True(oracleGroup.TotalPrice == 
                    oracleGroup.Trades.Sum(t => t.TotalTradePrice));


    // get the metaGroup and assert it exists and single
    SymbolTradeGroup metaGroup =
        symbolTradeGroups.Single(g => g.TheSymbol == Symbol.META);

    Assert.True(metaGroup.Trades.Count() == 2);
    Assert
        .True(metaGroup.TotalPrice == 
                  metaGroup.Trades.Sum(t => t.TotalTradePrice));

    // change the 
    oracleTrade1.TotalTradePrice = 5000;
    // assert that the aggregation's total price got updated
    Assert
        .True
        (oracleGroup.TotalPrice == 
              oracleGroup.Trades.Sum(t => t.TotalTradePrice));

    // add another oracle trade to the source collection
    var oracleTrade3 = Symbol.ORCL.CreateTrade(7000);
    sourceTradeCollection.Add(oracleTrade3);

    // make sure that number of traded without orcl group
    // is now 3 and that the total price of the aggregation
    // got also updated
    Assert.True(oracleGroup.Trades.Count() == 3);
    Assert
        .True(oracleGroup.TotalPrice == 
                   oracleGroup.Trades.Sum(t => t.TotalTradePrice));

    // add a trade for another instrument (e.g. Symbol.TSLA)
    // to the source collection
    var tslaTrade1 = Symbol.TSLA.CreateTrade(10);
    sourceTradeCollection.Add(tslaTrade1);

    // make sure that the number of groups is now 3:
    Assert.True(symbolTradeGroups.Count() == 3);

    // remove all Oracle trades from the source collection
    sourceTradeCollection.Remove(oracleTrade1);
    sourceTradeCollection.Remove(oracleTrade2);
    sourceTradeCollection.Remove(oracleTrade3);

    // make sure that the corresponding group is also removed
    Assert.True(symbolTradeGroups.Count() == 2);
    Assert.True
    (
        symbolTradeGroups
            .Where(group => group.TheSymbol == Symbol.ORCL)
            .Count() == 0);
}

This test uses an extra type SymbolTradeGroup representing a group of trades grouped by TheSymbol property:

public class SymbolTradeGroup : IDisposable
{
    // Dynamic Data group
    IGroup<Trade, int, Symbol> _group;

    // trades within the group
    public IEnumerable<Trade> Trades => _group.Cache.Items;

    // group key
    public Symbol TheSymbol => _group.Key;

    // sum of TotalTradePrice across all
    // the trades within the group
    public decimal TotalPrice { get; private set; }


    private IDisposable? _disposableSubscription;
    public void Dispose()
    {
        // destroy the group cache
        _group.Cache.Dispose();

        // remove the aggregation subscription
        _disposableSubscription?.Dispose();
        _disposableSubscription = null;
    }


    public SymbolTradeGroup(IGroup<Trade, int, Symbol> group)
    {
        _group = group;

        // set up the TotalPrice to be 
        // dynamically calculated when the group
        // or individual trades are changed
        _disposableSubscription =
            _group
                .Cache
                .Connect()
                .ToCollection()
                .Select(collection => 
                             collection.Sum(t => t.TotalTradePrice))
                .Subscribe(sum => this.TotalPrice = sum);
    }
}

Its TheSymbol property represents the common symbol (the grouping key). Its property TotalPrice is the sum of TotalTradePrice properties from each of its Trade objects. The TotalPrice will automatically update when the group content changes or when the individual Trade objects are updated.

Note, that SymbolTradeGroup depends on IGroup implementation from DD and is IDisposable because of that (whenever a group is removed the underlying DD functionality needs to be disposed of).

Now, let us dive into the GroupingTest() code.

The very beginning is exactly the same as in the previous sample - we create 4 trades, build an ObservableCollection<Trade> out of them and then obtain the IObservable<IChangeSet<Trade, int>> changeSetStream by calling ToObservableChangeSet(...) static extension method.

Then we do the grouping by TheSymbol property:

IObservable<IGroupChangeSet<Trade, int, Symbol>>
    groupedObservable =
        changeSetStream
            .AutoRefresh()
            .Group(t => t.TheSymbol);

Note, that AutoRefresh() will help to update the aggregated TotalPrice when TotalTradePrice changes on an individual trade object.

After grouping we empty Transform(...) extension method to convert each group into SymbolTradeGroup object:

var transformedGroups =
       groupedObservable.Transform(g => new SymbolTradeGroup(g));

Next, we define collection symbolTradeGroups to contain the groups, call Bind(...) extension method to get the groups into the collection and call Subscribe(...) to start pulling data:

using IDisposable subscription =
    transformedGroups
        .Bind(out symbolTradeGroups) // create and populate
        .DisposeMany() // make sure that if an item is removed
                       // from the collection, it is disposed
        .Subscribe();  // start the subscription

Note, the method DisposeMany() called before Subscribe() - it ensures that once an group is removed from the symbolTradeGroups collection, its Dispose() method is called.

After that we assert the number of groups, the TotalPrice value within each group and the content of each of the groups:

Assert.True(metaGroup.Trades.Count() == 2);
Assert
    .True(metaGroup.TotalPrice == 
             metaGroup.Trades.Sum(t => t.TotalTradePrice));

// change the 
oracleTrade1.TotalTradePrice = 5000;
// assert that the aggregation's total price got updated
Assert
    .True
    (oracleGroup.TotalPrice == oracleGroup.Trades.Sum(t => t.TotalTradePrice));

After that, we update some TotalTradePrice values for individual trades, add and remove some trades to sourceTradeCollection and after each operation verify that the number of the groups, their TotalPrice properties and trades are what they are supposed to be.

Introduction to Dynamic Data

Nick Polyak — Wed, 26 Feb 2025 18:42:30 +0000

What is DynamicData for?

DynamicData allows monitoring source collections and updating target collection in such a way that the updated collections satisfy certain conditions specified by the DynamicData operators.

The great feature of DynamicData is that the target collections pick up the source collection inserts, removes and updates and arrange the corresponding changes in the target collections to satisfy the same conditions.

This makes DynamicData very useful especially on the front end, allowing to manipulate the collections of non-visual objects, while the visual objects will be passively reflecting the changes.

Dynamic Data has been built by Roland Pheasant (apparently around 2014) by applying Reactive Extensions to monitor and update collections and since then consistently maintained and improved by him.

To the best of my knowledge, unlike Reactive Streams, DynamicData exists only in .NET universe (has not been translated to other languages yet).

Source Code Location

All the non-visual source code for this article is located under Intro to Dynamic Data project under NP.Samples repository. Clone or download the repository and start IntroToDynamicData.sln in Visual Studio 2022 to run the samples.

The samples are written as XUnit fixtures - for more on XUnit read Everything you need to know to create XUnit C# Tests in Visual Studio.

All samples for this article test methods (marked by [Fact] attribute) within SimpleExamples.cs file.

Introductory Sample

Our first sample demonstrates maintaining an output (target) collection of integers mimicking the input (source) collection, while including only even numbers, sorting them in descending order and taking their squares.

Our first example in located in file SimpleExamples.cs, test method SimpleExamples.IntTransformFilteringAndSorting().

Do not try to understand the whole code at once since I'll describe it almost line by line below.

[Fact]
public static void IntTransformFilteringAndSorting()
{
    // create source collection and populate
    // it with numbers 1 to 6.
    ObservableCollection<int> sourceInts = 
        new ObservableCollection<int>(Enumerable.Range(1,6));

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<int>> changeSetStream = 
        sourceInts.ToObservableChangeSet();

    // create the target collection
    IObservableCollection<int> targetInts = 
        new ObservableCollectionExtended<int>();

    IObservable<IChangeSet<int>> resultObservable =
        changeSetStream
            // filter only even ints
            .Filter(i => i % 2 == 0)
            // sort descending
            .Sort(SortExpressionComparer<int>.Descending(i => i))
            // transform to squares
            .Transform(i => i * i)
            // bind the target collection to update
            .Bind(targetInts);

    // now subscribe to start pulling data
    // using clause will dispose the subscription
    using IDisposable subscribeDisposable =
        resultObservable.Subscribe();

    // make sure the target collection is correct
    Assert.True(targetInts.SequenceEqual([6 * 6, 4 * 4, 2 * 2]));

    // remove 2 (even int) from source collection
    sourceInts.Remove(2);

    // make sure the target collection is correct
    Assert.True(targetInts.SequenceEqual([6 * 6, 4 * 4]));


    // insert 3 integers 20, 21 and 22 in the
    // source collection. Notice that only even numbers
    // 20 and 22 will influence the 
    // target collection.
    sourceInts.Insert(0, 20);
    sourceInts.Insert(1, 21);
    sourceInts.Insert(2, 22);

    // make sure the target collection is correct
    Assert.True(targetInts.SequenceEqual([22 * 22, 20 * 20, 6 * 6, 4 * 4]));

    // empty out the source collection
    sourceInts.Clear();

    // check that the target collection
    // has also been emptied out
    Assert.Empty(sourceInts);
}

First, we create a source collection and populate it with numbers 1 to 6:

ObservableCollection<int> sourceInts = 
    new ObservableCollection<int>(Enumerable.Range(1,6));

Type ObservableCollection<T> is part of C# language, not specific to DynamicData. It implements INotifyCollectionChanged and fires CollectionChanged event whose arguments contain information about the changes to the collection. From this information, DynamicData creates an observable stream of changes to the collection by calling ToObservableChangeSet() extension method:

IObservable<IChangeSet<int>> changeSetStream = 
     sourceInts.ToObservableChangeSet();

Then we define and create the target collection targetInts:

IObservableCollection<int> targetInts = 
    new ObservableCollectionExtended<int>();

ObservableCollectionExtended<T> is a DynamicData type - an improvement over ObservableCollection<T> allowing more control over when CollectionChanged events are fired.

Now we create an observable of collection changes we want by manipulating the input changeSetStream with DynamicData operators:

IObservable<IChangeSet<int>> resultObservable =
    changeSetStream
        // filter only even ints
        .Filter(i => i % 2 == 0)
        // sort descending
        .Sort(SortExpressionComparer<int>.Descending(i => i))
        // transform to squares
        .Transform(i => i * i)
        // bind the target collection to update
        .Bind(targetInts);

Note the DynamicData operators are named differently than LINQ and Rx operators. This is done on purpose to prevent the mix-ups with the corresponding Rx operators.

In particular

Filter(...) stands for Where(...)
Sort(...) is used instead of OrderBy(...)
Transform(...) instead of Select(...)

The 3 operators mean that we filter in only even integers, sort them by descending order and take their squares as outputs.

The most interesting is the operator Bind(...) to which you pass the output collection. It binds the target collection to update in compliance with the operators performed on the source collection changes once the subscription is made.

Note: nothing will happen until we call Subscribe() on the resulting IObservable<IChangeSet<int>> - similar to IObservable<T> in Rx.NET:

using IDisposable subscribeDisposable =
    resultObservable.Subscribe();

For the rest of the method, we use Assert.True(...) statements to make sure that the target collection's content is what we expect to be while modifying the source collection.

Dynamic Filter Demo

Our next small demo will show how the filter can be changed dynamically with the change being picked up immediately by the Dynamic Data. The test is located within the same file - SimpleExamples.cs and the test is a static method SimpleExamples.DynamicFilterTest():

[Fact]
public static void DynamicFilterTest()
{
    // create source collection and populate
    // it with numbers 1 to 6.
    ObservableCollection<int> sourceInts =
        new ObservableCollection<int>(Enumerable.Range(1, 6));

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<int>> changeSetStream =
        sourceInts.ToObservableChangeSet();

    // create the target collection
    IObservableCollection<int> targetInts =
        new ObservableCollectionExtended<int>();

    Subject<Func<int, bool>> filter = 
        new Subject<Func<int, bool>>();

    IObservable<IChangeSet<int>> resultObservable =
        changeSetStream
            // filter only even ints
            .Filter(filter)
            // sort descending
            .Sort(SortExpressionComparer<int>.Descending(i => i))
            // transform to squares
            .Transform(i => i * i)
            // bind the target collection to update
            .Bind(targetInts);

    // now subscribe to start pulling data
    // using clause will dispose the subscription
    using IDisposable subscribeDisposable =
        resultObservable.Subscribe();

    // add filter only after the subscription
    // filter in only even numbers
    filter.OnNext(i => i % 2 == 0);

    // make sure the target collection is correct
    Assert.True(targetInts.SequenceEqual([6 * 6, 4 * 4, 2 * 2]));

    // filter in only number divisible by 3
    filter.OnNext(i => i % 3 == 0);

    // make sure the target collection is correct
    Assert.True(targetInts.SequenceEqual([6 * 6, 3 * 3]));

    // add a value 9 (divisible by 3) to the source sequence.
    sourceInts.Add(9);

    Assert.True(targetInts.SequenceEqual([9 * 9, 6 * 6, 3 * 3]));
}

The test method similar to the previous one, so I'll be concentrating only on the differences.

In previous test method SimpleExamples.IntTransformFilteringAndSorting(), we passed the lambda expression of type Func<int, bool> to the Filter(...) method:

changeSetStream
    // filter only even ints
    .Filter(i => i % 2 == 0)
    ...

to filter in only even numbers.

Now, instead of a Func<int, bool> lambda expression, we shall pass IObservable<Func<int, bool>> - so that we could replace the filtering lambda expression at will.

To make it simple (changing the filtering lambda expression) we shall define the filter IObservable<Func<int, bool>> as Subject<Func<int, bool>> (remember from the previous article that subjects allow us not only to subscribe to Rx Streams, but also put the new data into them):

Subject<Func<int, bool>> filter = 
    new Subject<Func<int, bool>>();

IObservable<IChangeSet<int>> resultObservable =
    changeSetStream
        // filter only even ints
        .Filter(filter)
        ...

Note, that we only should start putting data into the filter after we subscribe to the collection changes (under the covers it also subscribes to the filter changes):

// now subscribe to start pulling data
// using clause will dispose the subscription
using IDisposable subscribeDisposable =
    resultObservable.Subscribe();

// add filter only after the subscription
// filter in only even numbers
filter.OnNext(i => i % 2 == 0);

// make sure the target collection is correct
Assert.True(targetInts.SequenceEqual([6 * 6, 4 * 4, 2 * 2]));

Next, let us change the filter - to filter in numbers divisible by 3 (and not even numbers as before):

// filter in only number divisible by 3
filter.OnNext(i => i % 3 == 0);

// make sure the target collection is correct
Assert.True(targetInts.SequenceEqual([6 * 6, 3 * 3]));

We immediately verify that the target sequence now contains square(6) and square(3).

Now you can play with adding and removing integers to and from the source and make sure that the target collection reflects the source changes.

Total Sum Aggregation Tests

Next we shall learn about summing up all numbers within a collection using Sum(...) operator. Note, there are several other aggregation operators built into Dynamic Data including Avg(...), Maxumum(...), Count() and Minimum(...) - they are self explanatory and all work in the same fashion as Sum(...).

The method demonstrating Sum(...) aggregation is SimpleExamples.TotalSubAggregationTest():

[Fact]
public static void TotalSumAggregationTest()
{
    // create source collection and populate
    // it with numbers 1 to 3.
    ObservableCollection<int> sourceInts =
        new ObservableCollection<int>(Enumerable.Range(1, 3));

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<int>> changeSetStream =
        sourceInts.ToObservableChangeSet();

    // create the collection sum observable
    IObservable<int> sumObservable = changeSetStream.Sum(i => i);

    // create a variable to hold the sum
    int sumResult = 0;

    // subscribe to changes in total sum
    // first time it fires after the subscription.
    using IDisposable disposableSubscription =
        sumObservable.Subscribe(result => sumResult = result);

    // 1 + 2 + 3 = 6
    Assert.Equal(6, sumResult);

    // add some numbers
    sourceInts.AddRange([5, 10, 15]);

    // 1 + 2 + 3 + 5 + 10 + 15 = 36
    Assert.Equal(36, sumResult);

    // remove all numbers
    sourceInts.Clear();

    // test that we obtain 0 as a result
    // of removing everything
    Assert.Equal(0, sumResult);
}

Note, that for the ease of figuring out the sum, I put only 1, 2 and 3 into the source collection, not numbers from 1 to 6 as in the previous samples.

Essentially the Sum(...) operator we use, transforms the IObservable<IChangeSet<int>> into IObservable<int> which represents the Rx stream of changes to the sum of all numbers within the collection:

IObservable<int> sumObservable = changeSetStream.Sum(i => i);

Every time the input collection changes, the sumObservable Rx stream emits the new sum of all its numbers.

Grouping Sample

Our next sample shows a grouping of a collection of integers by the remainder from the division by 2 (even numbers vs odd numbers):

[Fact]
public static void GroupingTest()
{
    // create source collection and populate
    // it with numbers 1, 3 and 5 (at this point only odd numbers).
    ObservableCollection<int> sourceInts =
        new ObservableCollection<int>([1, 3, 5]);

    // create stream of IChange<int> parameters
    // from the source collection
    IObservable<IChangeSet<int>> changeSetStream =
        sourceInts.ToObservableChangeSet();

    // group the numbers using GroupWithImmutableState operator
    // by the remainder from division by 2 (odd vs even)
    IObservable<IChangeSet<DD.List.IGrouping<int, int>>> 
        groupedObservable =
            changeSetStream.GroupWithImmutableState(i => i % 2);

    // create the grouped collection
    using IObservableList<DD.List.IGrouping<int, int>> 
        groupedCollection =
            groupedObservable
                .AsObservableList();

    // grouped collection only has one group of odd numbers
    // since there are only odd number in the source collection:
    Assert.True(groupedCollection.Count == 1);

    // get odd group
    DD.List.IGrouping<int, int> oddGroup =
        groupedCollection.Items.Single(grouping => grouping.Key == 1);

    Assert.True(oddGroup.Items.SequenceEqual([1, 3, 5]));

    // add even numbers 2, 4, 6 to the source collection
    sourceInts.AddRange([2, 4, 6]);

    // now there should be two groups of numbers - odd and even
    Assert.True(groupedCollection.Count == 2);

    // get odd group
    oddGroup = 
        groupedCollection.Items.Single(grouping => grouping.Key == 1);

    // check that the numbers in the odd group are [1, 3, 5]
    Assert.True(oddGroup.Items.SequenceEqual([1, 3, 5]));

    // get event group
    DD.List.IGrouping<int, int> evenGroup =
        groupedCollection.Items.Single(grouping => grouping.Key == 0);

    // check that the numbers in the even group are [2, 4, 6]
    Assert.True(evenGroup.Items.SequenceEqual([2, 4, 6]));
}

First we populate the source collection only with odd number (1, 3 and 5):

ObservableCollection<int> sourceInts =
        new ObservableCollection<int>([1, 3, 5]);

We use GroupWithImmutableState operator to group the source collection by the whether the number is odd or even (remainder from the division by 2):

IObservable<IChangeSet<DD.List.IGrouping<int, int>>> 
    groupedObservable =
        changeSetStream.GroupWithImmutableState(i => i % 2);

Note that we had set using DD = DynamicData; to shorten the line.

Next, we transform the Rx observable stream of changes into IObservableList<DD.List.IGrouping<int, int>> by using AsObservableList operator:

using IObservableList<DD.List.IGrouping<int, int>> 
    groupedCollection =
        groupedObservable
            .AsObservableList();

AsObservableList() calls Subscribe() underneath and the resulting IObserabableList has to be disposed. This is why we employ using keyword.

Next we check that the result has only one group (of odd numbers) and that it consists of number 1, 3 and 5:

Assert.True(groupedCollection.Count == 1);

// get odd group
DD.List.IGrouping<int, int> oddGroup =
    groupedCollection.Items.Single(grouping => grouping.Key == 1);

Assert.True(oddGroup.Items.SequenceEqual([1, 3, 5]));

Now, let us add even numbers 2, 4 and 6 to the source collection:

sourceInts.AddRange([2, 4, 6]);

The resulting groupedCollection will now consist of two groups (one containing odd and the other even numbers):

Assert.True(groupedCollection.Count == 2);

The odd group will contain numbers 1, 3 and 5, while the even group will contain numbers 2, 4 and 6:

oddGroup = 
    groupedCollection.Items.Single(grouping => grouping.Key == 1);

// check that the numbers in the odd group are [1, 3, 5]
Assert.True(oddGroup.Items.SequenceEqual([1, 3, 5]));

// get event group
DD.List.IGrouping<int, int> evenGroup =
    groupedCollection.Items.Single(grouping => grouping.Key == 0);

// check that the numbers in the even group are [2, 4, 6]
Assert.True(evenGroup.Items.SequenceEqual([2, 4, 6]));

Grouping with Binding and Aggregation Sample

Previous sample produces an ObservableList of groups. This sample will produce an observable collection of objects of IntsGroup type which is a wrapper around Dynamic Data's IGroup<int, int> type with added Sum property representing the sum of all integers within the group:

public class IntsGroup : IDisposable
{
    // Dynamic Data group
    public IGroup<int, int> Group { get; }

    // group key
    public int GroupKey => Group.GroupKey;

    // collection of integers 
    // within the group
    public IEnumerable<int> Values => Group.List.Items;

    // represents sum of all integers within the group
    public int Sum { get; set; }

    // handle to dispose the subscription
    private IDisposable? _aggregationDisposable = null;

    // pass DynamicData group of integers.
    public IntsGroup(IGroup<int, int> group)
    {
        Group = group;

        // every time the collection changes,
        // change the Sum property to reflect 
        // the sub of interges within the collection
        _aggregationDisposable =
            group
                .List
                .Connect()
                .ToCollection()
                .Select(collection => collection.Sum())
                .Subscribe(sum => Sum = sum);
    }

    public void Dispose()
    {
        if (_aggregationDisposable != null)
        {
            // need to dispose the subscription 
            _aggregationDisposable?.Dispose();

            _aggregationDisposable = null;
        }
    }
}

To make sure that Sum property changes every time collection integers are added or removed from the group, we subscribe to the group's List changes within the constructor:

public IntsGroup(IGroup<int, int> group)
{
    Group = group;

    // every time the collection changes,
    // change the Sum property to reflect 
    // the sub of interges within the collection
    _aggregationDisposable =
        group
            .List
            .Connect()
            .ToCollection()
            .Select(collection => collection.Sum())
            .Subscribe(sum => Sum = sum);
}

Here is the code for the test:

[Fact]
public static void GroupingWithBindingAndAggregationTest()
{
    // create source collection and populate
    // it with numbers 1 to 6.
    ObservableCollection<int> source =
        new ObservableCollection<int>(Enumerable.Range(1, 6));

    // create a stream of IChangeSet<int> parameters
    // from the source collection
    IObservable<IChangeSet<int>> sourceChangeObservable =
        source.ToObservableChangeSet();

    // create a collection of IntsGroup objects
    // representing the grouped integers
    IObservableCollection<IntsGroup> groupedInts = 
        new ObservableCollectionExtended<IntsGroup>();

    // using clause is to remove the 
    // subscription by the end of the method.
    using IDisposable disposableSubscription =
        sourceChangeObservable

            // group by remainder from division by 2
            // (even vs odd numbers)
            .GroupOn(i => i % 2)

            // transform each group
            // to IntsGroup object
            .Transform(g => new IntsGroup(g))

            // bind to groupedInts collection
            .Bind(groupedInts)

            // make sure once an IntsGroup
            // object is removed from the collection,
            // it is also deleted
            .DisposeMany()

            // Subscribe - start pulling
            // the changes
            .Subscribe();

    // get the even group of integers
    IntsGroup evenGroup =
        groupedInts.FirstOrDefault(g => g.GroupKey == 0)!;

    // get the odd group of integers
    IntsGroup oddGroup =
        groupedInts.FirstOrDefault(g => g.GroupKey == 1)!;

    // assert that the even group has 2, 4 and 6 values in it
    Assert.True(evenGroup.Values.SequenceEqual([2, 4, 6]));

    // assert that the Sum property of the even group
    // equals to the sum of 2, 4 and 6
    Assert.Equal(2 + 4 + 6, evenGroup.Sum);

    // assert that the odd group has 1, 3 and 5 values in it
    Assert.True(oddGroup.Values.SequenceEqual([1, 3, 5]));

    // assert that the odd group's Sum property 
    // equals to the sum of 1, 3 and 5
    Assert.Equal(1 + 3 + 5, oddGroup.Sum);

    // add odd number 7 to the source 
    source.Add(7);

    // Assert that the even group remained the same
    Assert.True(evenGroup.Values.SequenceEqual([2, 4, 6]));

    // Assert that even group's Sum remains the same
    Assert.Equal(2 + 4 + 6, evenGroup.Sum);

    // make sure that odd group added number 7 at the end
    Assert.True(oddGroup.Values.SequenceEqual([1, 3, 5, 7]));

    // Assert that the odd group's Sum property 
    // equals to the sum of 1, 3, 5 and 7
    Assert.Equal(1 + 3 + 5 + 7, oddGroup.Sum);
}

Note that we are using Transform(...) and Bind(...) methods to populate the observable collection IObservableCollection<IntsGroup> groupedInts:

// using clause is to remove the 
// subscription by the end of the method.
using IDisposable disposableSubscription =
    sourceChangeObservable

        // group by remainder from division by 2
        // (even vs odd numbers)
        .GroupOn(i => i % 2)

        // transform each group
        // to IntsGroup object
        .Transform(g => new IntsGroup(g))

        // bind to groupedInts collection
        .Bind(groupedInts)

        // make sure once an IntsGroup
        // object is removed from the collection,
        // it is also deleted
        .DisposeMany()

        // Subscribe - start pulling
        // the changes
        .Subscribe();

Then we assert the correct collection content and the correct sum of the items within the even and odd groups:

// get the even group of integers
IntsGroup evenGroup =
    groupedInts.FirstOrDefault(g => g.GroupKey == 0)!;

// get the odd group of integers
IntsGroup oddGroup =
    groupedInts.FirstOrDefault(g => g.GroupKey == 1)!;

// assert that the even group has 2, 4 and 6 values in it
Assert.True(evenGroup.Values.SequenceEqual([2, 4, 6]));

// assert that the Sum property of the even group
// equals to the sum of 2, 4 and 6
Assert.Equal(2 + 4 + 6, evenGroup.Sum);

// assert that the odd group has 1, 3 and 5 values in it
Assert.True(oddGroup.Values.SequenceEqual([1, 3, 5]));

// assert that the odd group's Sum property 
// equals to the sum of 1, 3 and 5
Assert.Equal(1 + 3 + 5, oddGroup.Sum);

Then we add one more (odd) number 7 to the source collection and re-test that the even group and its sum remained the same while the odd group and its sum changed correspondingly:

// add odd number 7 to the source 
source.Add(7);

// Assert that the even group remained the same
Assert.True(evenGroup.Values.SequenceEqual([2, 4, 6]));

// Assert that even group's Sum remains the same
Assert.Equal(2 + 4 + 6, evenGroup.Sum);

// make sure that odd group added number 7 at the end
Assert.True(oddGroup.Values.SequenceEqual([1, 3, 5, 7]));

// Assert that the odd group's Sum property 
// equals to the sum of 1, 3, 5 and 7
Assert.Equal(1 + 3 + 5 + 7, oddGroup.Sum);

[Boost]

Nick Polyak — Fri, 14 Feb 2025 15:30:47 +0000

Introduction to Rx Observable Streams and Dynamic Data

Nick Polyak ・ Feb 7

#csharp #reactiveextensions #dynamicdata #reactiveui

[Boost]

Nick Polyak — Sun, 09 Feb 2025 15:56:28 +0000

Using XUnit Test Framework for Testing C# Code under Visual Studio

Nick Polyak ・ Feb 9

#csharp #xunit #testing

Everything you need to know to create XUnit C# Tests in Visual Studio

Nick Polyak — Sun, 09 Feb 2025 01:12:01 +0000

This blog post will save you time and effort by providing you with all the information you need for creating software tests with XUnit, so that you do not have to collect this information from disparate sources. It details creating the test projects, specifies what nuget packages are required, how to write and run the tests and how to send information to the test console.

For an article that treats Test Driven Development from the architectural point of view and in much more detail, please see Test Driven Development Process with XUnit.

Most console programs have single Main() method to run. Testing frameworks allow you to run multiple methods called Tests. Each such method usually correspond to testing a certain feature.

XUnit is one of the most popular .NET testing frameworks. Its test methods are marked with [Fact] or [Theory] attribute ([Theory] allows you to run the same method multiple times with different arguments).

To make it possible to run [Fact]s and [Theory]s in Visual Studio you create the your testing project as a class library and add xunit nuget package to it.

To make it possible to debug [Fact]s and [Theory]s in Visual Studio you have to add two more nuget packages: Microsoft.NET.Test.SDK and xunit.runner.visualstudio:

To run or debug the tests you should open the test explorer from the Visual Studio:

Then find the test you want to run in it and press Run or Debug button:

If you use debug, the execution will stop once a breakpoint is hit.

The code for the very simple XUnit samples is located under XUnitSample project on github.com.

File XUnitTests.cs contains two tests of built-in .NET method string.Split() - one to demonstrate [Fact] and another - [Theory].

Here is the Fact sample:

// fact sample
[Fact]
public void TestSplit()
{
    string str = "Hello World";
    string[] split = str.Split(' ', StringSplitOptions.RemoveEmptyEntries);

    // assert that the split is a collection of two strings:
    // "Hello" and "World"
    Assert.True(split.SequenceEqual(["Hello", "World"]));
}

We split string "Hello World" into an array containing two strings - "Hello" and "World" and we assert that the result is correct.

Here is the code for [Theory] test:

/// <summary>
/// Run TestSplits(...) methods for every [InlineData] 
/// providing various arguments.
/// </summary>
[Theory]
[InlineData("Hello", new string[] { "Hello" })]
[InlineData("Hello World", new string[] { "Hello", "World" })]
[InlineData("Hello Great World", new string[] { "Hello", "Great", "World" })]
public void TestSplits(string strToSplit, string[] expectedSplit)
{
    string[] split = strToSplit.Split(' ', StringSplitOptions.RemoveEmptyEntries);

    Assert.True(split.SequenceEqual(expectedSplit));
}

TestSplits(...) method has two arguments:

stringToSplit- is the original string to be splitted
expectedSplit - argument containing the array with the expected result of the split.

The [InlineData] attributes provide both arguments to the TestsSplits(...) method. Since there are three [InlineData] attributes - the TestsSplits(...) method will be run three times (for each set of arguments).

There is also XUniTestWithPrinting.cs file. It demonstrate how to create tests that print information you want to the test console (the usual Console.WriteLine(...) methods will not work). All you need to do is to provide a constructor taking an argument of ITestOutputHelper type and assign the value of this argument to a property or a field of the class:

public class XUnitTestWithPrinting
{
    private ITestOutputHelper _xunitOutput;

    /// <summary>
    /// pass a test console writer for the test 
    /// (XUnit will provide the correct value itself)
    /// </summary>
    /// <param name="xunitOutput"></param>
    public XUnitTestWithPrinting(ITestOutputHelper xunitOutput)
    {
        _xunitOutput = xunitOutput;
    }
    ...
}

Then you can use this property of field to print to the test console as it is done in XUnitTestWithPrinting.TestSplit() method:

[Fact]
public void TestSplit()
{
    string str = "Hello World";
    string[] split = str.Split(' ', StringSplitOptions.RemoveEmptyEntries);

    Assert.True(split.SequenceEqual(["Hello", "World"]));

    foreach (string part in split)
    {
        // print each part to Test Console
        _xunitOutput.WriteLine(part);
    }
}

The results will be shown in the TestExplorer window on the right:

Using Rx Observable Streams and Dynamic Data Patterns to build Applications Clean and Fast (Introduction)

Nick Polyak — Fri, 07 Feb 2025 21:27:37 +0000

What this article is about

This is an introductory article on the fascinating subject of mastering two very powerful but underused, misused and often not well understood frameworks: Reactive Extensions and Dynamic-Data. More in-depth articles dedicated to these two frameworks will be coming soon.

Using both packages properly can greatly simplify and speed up designing and building real world applications, both backend (or middle tier) and the client side.

Reactive Extensions provide a way to easily manipulate streams of data by applying LINQ-like operators to them.

Dynamic Data allows observing collections' changes as streams and changing/combining streams using LINQ-like operators to create and update collections. This is very important, especially on the front end to create displayable collections. For example, using Dynamic Data one can create a data-grid with almost all the Virtualization, Filtering, Sorting and even Grouping capability provided on the non-visual collections (view models) with the visual DataGrid passively reflecting the non-visual objects manipulated by the Dynamic Data.

The reason Reactive Extensions and Dynamic Data frameworks are not used as much as they could have been is because of the lack of well written documentation on how to use them. This is precisely the purpose of this and the coming articles - to describe the correct usage patterns for Reactive Extensions and Dynamic Data.

All code samples for his series are written in C#/.NET 9 so some familiarity with C# and especially with its LINQ operators is assumed.

Introduction to Reactive Extensions (Rx)

Reactive Extensions provide a way of manipulating data coming in streams. Each stream usually emits data of the same type. Reactive Extensions provide numerous set of operators that allow manipulating the streaming data in various ways, including

Transforming
Filtering
Sorting
Combining multiple streams into one
Grouping

and many others.

In fact a stream can be considered a dual of a collection and all the LINQ Collection operators have their dual operators for Rx streams. I will not get much more into duality here - duality was important for discovering the concept but not so important for using and understanding it.

Unlike collection, streams also have a time dimension. Therefore, many operators that work on Rx streams are built to work in time and do not have their corresponding duals in LINQ. As a result, Rx streams have more operators than their collection counterparts.

Rx was first created by Microsoft around 2010-2011, as part of their .NET framework and was called Rx.NET. Since then, Reactive Extensions were translated into many other languages including JavaScript (RxJs), Java (RxJava), Python (RxPY) and many others.

Reactive Extensions, especially in combination with Dynamic Data, has a great potential to make software development considerably easier and faster (similar to what LINQ operators did for collections, but more so, since streams can be used in many more scenarios than collections and I'll expand on it further down). Also, note that the streams are the best way of communications between the backend (or middle tier) software and the front end (Web, Mobile or Desktop).

Despite these unique advantages, Reactive Extensions have not lived up to its potential. Many people encountered Rx, did a bit of a coding with it, but very few of them can be called Reactive Extensions experts. As mentioned above, the main reason for that is the lack of good documentation explaining how to use Reactive Extensions.

You can find a lot of Rx documentation on the web, for example ReactiveX or Introduction to Rx, but these on-line books are rather handbooks - detailing each Rx.NET operator and describing how they work in detail. They are very valuable when one needs to figure out how to use a specific Rx operator, but a bit boring to read (as any handbook would be).

There is also a wealth of samples e.g. reactive 101 samples. While these samples also provide a good reference guide, they do not highlight the main Rx concepts.

Instead of writing another handbook, my primary purpose is to show how to use Reactive Extensions correctly to achieve your goal of building real life applications clean and fast.

Introduction to Dynamic Data

Dynamic Data has been built by Roland Pheasant (apparently around 2014) by applying Rx.NET to monitor and update collections.

Using Dynamic Data one can do some really crazy things with collections, for example Create a collection B that monitors some other collection A for updates, removes and inserts so that each item within B corresponds to an item within A as long as some condition on the item is satisfies certain condition (filtering).

Using Dynamic Data you can apply many other operations on collections also (e.g. transformations, sorting, grouping, combing elements from different collections into one collection).

Dynamic Data very fascinating and gives you a lot of power especially when it comes to displaying collection items in various ways.

Documentation for Dynamic Data is also somewhat lacking. There are some blog entries by Roland Pheasant at Dynamic Data, some good explanations at reactivemarbles/dynamic-data, many responses to questions on Stackoverflow (c# and dynamic-data tags) and there are some samples on github, e.g. at DynamicData.Snippets and Dynamic.Trader.

Rx.NET Streams

Stream is an object that emits data. Usually single stream emits data of the same type within the same thread or async context multiple times with various time intervals between the emissions - time intervals do not have to be the same. Some streams can also be empty or have only a single data emission.

Rx Streams can be observed (subscribed to). If a stream is observed, each data emission will result in an action (event) fired with the emitted data being passed as the single parameter to that action.

In Rx.NET each stream is represented by IObservable<T> type (from System.Runtime.dll which means that this interface is basic part of C# language). Type argument T is the type of data being emitted. Here is the definition of IObservable<T> interface

// Summary:
//     Defines a provider for push-based notification.
// Type parameters:
//   T:
//     The object that provides notification information.
public interface IObservable<out T>
{
    IDisposable Subscribe(IObserver<T> observer);
}

The interface is very simple - it has only one method - Subscribe(...) that takes an object IObserver<T> observer as its only argument and returns an IDisposable.

IObserver<T> interface is also defined in System.Runtime.dll. It has three methods:

// Summary:
//     Provides a mechanism for receiving push-based notifications.
// Type parameters:
//   T:
//     The object that provides notification information.
public interface IObserver<in T>
{
    void OnNext(T value);
    void OnCompleted();
    void OnError(Exception error);
}

Each of these 3 methods will be called on an IObserver<T> which is subscribed to (observes) the IObservable<T> stream under the following conditions:

OnNext(T value) will be called when next data is emitted by the observed stream (potentially the stream might be empty and it might never be called).
OnCompleted() will be called when the stream signals that it has completed properly - no more data will be coming through it. Some streams might never signal completion so this method, might not ever be called either.
OnError(Exception error) will be called only if something went wrong with the stream and the stream signals that it is broken.

No data will be coming through the stream after it signaled completion or error.

Without Subscribe(...) being called on it, IObservable<T> stream is doing nothing - compare it to a shutdown water faucet. When one or several Subscribe(...) methods are called on an observable stream - the data might start flowing - dripping through the stream.

Disposing the returned IDisposable object will unsubscribe.

For those interested in duality - IObservable<T> is dual to IEnumerable<T> and IObserver<T> is dual to IEnumerator<T> - a very rarely used interface serving as an iterator through an IEnumerable<T> collection. Instead of using and implementing IEnumerator<T>, we use extention methods defined in static class System.Linq.Enumerable to sort, filter, transform or combine several IEnumerable<T> collection objects into the Enumerable collection we want.

In the similar fashion IObserver<T> is almost never used or implemented outside of the Rx packages. Instead various observable streams are sorted, filtered, transformed or combined into other stream(s) in many different ways by various extension methods many of which are defined in static class System.Reactive.Linq.Observable.

Reactive Stream Sources

Using Streams of Remote Data

The most natural and fastest ways of receiving streams - getting remote data as stream(s) (as opposed to the older request/reply paradigm).

Many modern communications packages supports streaming, for example:

gRPC (Google RPC) and its simplified version grpc-web - one of the most popular and fastest middleware protocols.
SignalR - one of the greatest and most powerful middleware packages - unfortunately, it requires .NET backend.
Web Sockets
Akka/Akka.NET
Kafka

Here is a communication schema of a multi-tier application:

The backend consists of databases and live streams (containing and transporting, e.g. financial or other data).

IObservable<T> Stream Adapter is part of the application that translates the streaming data into IObservable<T> streams (or streams that are easy to convert to IObservable<T>). Note that it can be factored out into e.g. a middle tier - as in the figure above - or it can be part of each client application.

This way of producing streams requires several tiers (parts of the application) communicating with each other. I will write a detailed article about it in the future as part of these series, but for now I shall provide other, more simple examples of creating Rx Streams.

Using System.Reactive and DynamicData Functionality to Create Streams

All the samples for this section are provided as XUnit [Fact]s in CreatingAndManipulatingRxStreams project, which is part of NP.Samples github public repository.

To run or debug the samples, download or clone the repository onto you local machine, cd to RxStream/CreatingAndManipulatingRxStreams folder and open CreatingAndManipulatingRxStreams.sln solution in Visual Studio 2022.

For more on running and debuggin XUnit projects, please take a look at Everything you need to know to create XUnit C# Tests.

Note that the project depends on DynamicData package which in turn depends on System.Reactive package so that we get full access to both Rx.Net and DynamicData functionality.

There is only a dependency on System.Linq.Async package, but only one test demonstrating converting IAsyncEnumerable<T> to IObservable<T> depends on it.

Converting Collection into `IObservable<T>`

Here we demo turning IEnumerable<T> collection into IObservable<T> streams, by using Observable.ToObservable(...) extension method.

The corresponding sample is locate within file StreamFromCollection.cs. Take a look at the static [Fact] method TestStreamFromCollection():

[Fact]
public static void TestStreamFromCollection()
{
    // source collection consists of int numbers 1 to 10
    int[] sourceCollection1_10 = 
                  Enumerable.Range(1, 10).ToArray();

    IObservable<int> sourceObservable = 
                        sourceCollection1_10.ToObservable();

    // observable emitting squares for only even numbers
    // out of the source
    IObservable<int> evenSquaresObservable =
            sourceObservable
            .Where(i => i % 2 == 0) // filter only even numbers
            .Select(i => i * i);    // transform i to square(i)


    // create result collection
    var resultEvenSquaresCollection = new List<int>();

    // populate result collection by subscribing
    // to it.
    // Note that before subscription, the observables
    // however complex they are do not consume any
    // heap resources and cannot result in a memory leak.
    // Only subscription needs to be dispose of 
    // (which we do automatically by providing 
    //  the 'using' clause).
    using var subscriptionDisposable =
        evenSquaresObservable
            .Subscribe(i => resultEvenSquaresCollection.Add(i));

    // create collection of squares of even positive ints
    // smaller or equal to 10 to compare the
    // result with
    int[] expectedResultsCollection =
        [2 * 2, 4 * 4, 6 * 6, 8 * 8, 10 * 10];

    // assert that the result and expectedResults collections
    // are the same.
    Assert.True
    (
         resultEvenSquaresCollection
              .SequenceEqual(expectedResultsCollection)
    );
}

Since this is our first Rx Stream example, I'll explain it almost line by line.

The following line creates an array containing int numbers from 1 to 10 as the sourceCollection to be turned into an observable.

int[] sourceCollection1_10 = Enumerable.Range(1, 10).ToArray();

We create an observable for every number within the collection by calling static extension method Observable.ToObservable(...) on it:

IObservable<int> sourceObservable = 
                     sourceCollection1_10.ToObservable();

We use the powerful composition LINQ-like operators for Rx Streams to filter in only even numbers of the Stream and transform the filtered even numbers into their squares:

IObservable<int> evenSquaresObservable =
       sourceObservable
        .Where(i => i % 2 == 0) // filter only even numbers
        .Select(i => i * i);    // transform i to square(i)

Note that since there has been no subscription made so far, nothing is happening, the stream is not operating and no resources need to be freed or disposed.

Next we'll create collection of results, we want to populate within our subscription method and we also subscribe to the observable to populate this collection:

List<int> resultEvenSquaresCollection = new List<int>();
using var subscriptionDisposable =
    evenSquaresObservable
        .Subscribe(i => resultEvenSquaresCollection.Add(i));

Within our Observable.Subscribe(...) extension method, we add each observed result to the resultEvenSquaresCollection. Subscription returns an IDisposable object which has to be disposed in order to prevent memory leaks (this is why we employ the using clause here). In more complex examples, disposing a subscription can also serve for unsubscribing from the stream.

The rest of the code merely tests that we indeed got a collection of squares of even positive integers less or equal to 10.

Observable.Subscribe(...) extension method has several overrides. We used the one that takes Action<T> onNext argument that fires when new data is emitted. We call also use other overrides that use Action onCompleted and Action<Exception> onError arguments as well.

For example we can define bool completed = false variable before the calling Subscribe(...) and then call:

bool completed = false;
using var subscriptionDisposable =
    evenSquaresObservable
        .Subscribe
         (
              i => resultEvenSquaresCollection.Add(i), 
              () => completed = true
         );

Assert.True(completed);

After observable stream is run, the completed flag will be flipped to true.

Another thing that could have been done differently is we did not need to create sourceObservable and evenSquaresObservable - I created them purely for clarity. In fact we could have done creating observable, performing filtering and transformation on it and subscribing within the same long C# statement:

using var subscriptionDisposable =
    sourceCollection1_10
        .ToObservable()
        .Where(i => i % 2 == 0)
        .Select(i => i * i)
        .Subscribe(i => resultEvenSquaresCollection.Add(i));

Another test TestTimeSpannedStreamFromCollection() from the same file shows how to create a 'real' (spaced in time) stream from the same collection by combining it with a Time Spanned stream using Zip(...) operator:

[Fact]
public static async Task TestTimeSpannedStreamFromCollection()
{
    // source collection consists of int numbers 1 to 10
    int[] sourceCollection1_10 = 
            Enumerable.Range(1, 10).ToArray();

    IObservable<int> evenSquaresObservable =
                        sourceCollection1_10
                            .ToObservable()
                            .Where(i => i % 2 == 0)
                            .Select(i => i * i);

    // create an infinite stream emitting 
    // empty data (Unit.Default stands for empty data)
    // every second
    IObservable<Unit> eachSecondObservable =
        Observable.Interval(TimeSpan.FromSeconds(1))
                    .Select(_ => Unit.Default);

    // Zip operator matches the data emitted from
    // two or more observable streams - 1st to 1st, 2nd to 2nd
    // etc.
    // Emission number N happens after EACH one of its
    // observable streams produces N-th emission. 
    //
    // The Zipped stream completes when ANY of its 
    // observable streams complete.
    IObservable<int> timeSpannedEvenSquaresObservable =
        evenSquaresObservable
            .Zip
            (
                eachSecondObservable,

                // we only need values from 
                // the first sequence - evenSquaresObservable
                // From the second sequence 
                // (eachSecondObservable) we only 
                // need time of emission
                (i, unit) => i
            );

    var resultEvenSquaresCollection = new List<int>();

    bool completed = false;
    using var subscriptionDisposable =
                    timeSpannedEvenSquaresObservable
                    .Subscribe
                    (
                        i => resultEvenSquaresCollection.Add(i),
                        () => completed = true
                    );

    // delay the testing until completed flag is 
    // switched to true
    while (! completed)
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
    }

    int[] expectedResultsCollection =
        [2 * 2, 4 * 4, 6 * 6, 8 * 8, 10 * 10];

    Assert.True
    (
        resultEvenSquaresCollection
           .SequenceEqual(expectedResultsCollection)
    );
}

Note that TestTimeSpannedStreamFromCollection() is async method and we wait for the completed flag to become true until we check the assertion. Running of this test should take more than 5 seconds.

And this is another example highlighting the great powers of Rx composition. In order to span out the emissions of the values of the collection we use Zip(...) operator to combine the evenSquaresObservable with another stream spanned in time. This (other) stream is created by using Observable.Interval(...) method:

IObservable<Unit> eachSecondObservable =
    Observable.Interval(TimeSpan.FromSeconds(1))
              .Select(_ => Unit.Default);

The eachSecondObservable emits data of type Unit.Default every second and it never sends the completion signal (it is potentially an infinite stream). Type Unit and its only value Unit.Default is used in Rx, when the emission data is not important (only emission timing is). In a sense it matches a C# event that takes no data (of type Action).

Then we combine these two streams (evenSquaresObservable and eachSecondObservable) into one by using Observable.Zip(...) operator:

IObservable<int> timeSpannedEvenSquaresObservable =
    evenSquaresObservable
        .Zip
        (
            eachSecondObservable,

            // we only need values from 
            // the first sequence - evenSquaresObservable
            (i, unit) => i
        );

Observable.Zip(...) operator combines two or more streams into one stream, by combining the corresponding data items (1st to 1st, 2nd to 2nd, ... , N-th to N-th, an so on).

The N-th item of the resulting stream is emitted only after N-th items are emitted in every stream it combines.

The resulting stream is completed when any of its source streams is completed.

In a sense we marry our evenSquaresObservable stream obtained from the collection and eachSecondObservable obtained from Observable.Interval(...) method taking emitted data from the first stream and emission time from the second.

IAsyncEnumerable to Stream

IAsyncEnumerable<T> is a different (lower level) way of presenting a stream built into C# language. IAsyncEnumerable<T> is especially important because gRPC and grpc-web streams are translated into IAsyncEnumerable<T> in C# and converting it into an IObservable<T> allows employing the powerful set on Rx Operators on it.

The demo of IAsyncEnumerable<T> converted into an IObservable<T> stream is located within StreamFromAsyncEnumerable.cs file.

The static method GenerateAsyncEnumerable() generates IAsyncEnumerable<T> stream:

// returns an IAsyncEnumerable stream
// of numbers from 1 to 6; each number
// takes one second to emit.
public static async IAsyncEnumerable<int> 
    GenerateAsyncEnumerable()
{
    for (int i = 1; i <= 6; i++)
    {
        // one second delay
        await Task.Delay(TimeSpan.FromSeconds(1));

        yield return i;
    }
}

It emits an integers from 1 to 6 every second.

Method TestStreamFromAsyncEnumerable() converts IAsyncEnumerable<T> into IObservable<T>, filters only even number and transforms them into their squares (same as in the previous sample) and then tests the result using Assert statement:

[Fact]
public static async Task TestStreamFromAsyncEnumerable()
{
    // Get asyncEnumerable
    IAsyncEnumerable<int> asyncEnumerable = 
        GenerateAsyncEnumerable();

    // convert to an observable
    // stream
    IObservable<int> observable = 
        asyncEnumerable.ToObservable();

    // create a result list to be populated
    // within the observable's Subscription
    var resultEvenSquaresCollection = new List<int>();

    // define completed flag
    bool completed = false;

    // subscribe to receive
    // squares of even numbers
    using var subscribeDisposable =
        observable

            // filter in only even numbers
            .Where(i => i % 2 == 0) 

            // transform i to square(i)
            .Select(i => i * i)
            .Subscribe
            (
                onNext:
                    i => resultEvenSquaresCollection.Add(i),

                onCompleted:
                    () => completed = true
            );

    // delay the testing until completed flag is 
    // switched to true
    while (!completed)
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
    }

    // assert that the result as expected
    Assert.True
    (
        resultEvenSquaresCollection
            .SequenceEqual([2 * 2, 4 * 4, 6 * 6])
    );
}

The only method unique to this test, demonstrated here, is AsyncEnumerable.ToObservable(...) static extension method defined within System.Linq.Async.dll to convert IAsyncEnumerable<T> into IObservable<T> stream.

Using Subjects for Creating Observable Streams.

Subjects are special Rx observables that also implement three methods allowing to push data into them:

OnNext(T data) - for pushing emitted data
OnCompleted() - for signaling the end of the stream
OnError(Exception e) for signaling a stream error

In fact ISubject<T> implements both IObservable<T> (for using it as a an observable stream source) and IObserver<T> (for populating it with data).

Subjects are great

For tests (since you have full control over the data you want to populate it with and completion/error behavior)
For creating a stream from some non Rx Stream data.
For using them to initialize auxiliary observable streams to facilitate stream composition.

Subject tests are located within StreamFromSubject.cs file.

Take a look at TestStreamFromSubject() method:

[Fact]
public static void TestStreamFromSubject()
{
    ISubject<int> subject = new Subject<int>();

    var resultEvenSquaresCollection = new List<int>();
    bool completed = false;
    string exceptionMessage = null;

    using var subscriptionDisposable =
        subject
            .Where(i => i % 2 == 0) // filter only even
            .Select(i => i * i) // get squares
            .Subscribe
            (
                onNext: i => resultEvenSquaresCollection.Add(i), 
                onCompleted: () => completed = true,
                onError: exception => 
                            exceptionMessage = exception.Message
            );

    // push data into the stream
    // by using OnNext() method
    subject.OnNext(1);
    subject.OnNext(2);
    subject.OnNext(3);
    subject.OnNext(4);

    // completion signal has not been sent yet
    Assert.False(completed);    

    // no exception occurred yet
    Assert.Null(exceptionMessage);

    Assert.True
    (
        resultEvenSquaresCollection
            .SequenceEqual([2 * 2, 4 * 4])
    );

    // add another even number
    subject.OnNext(10);

    // verify the output of the stream
    Assert.True
    (
        resultEvenSquaresCollection
            .SequenceEqual([2 * 2, 4 * 4, 10 * 10])
    );

    // no completion yet
    Assert.False(completed);

    // signal source stream completion
    subject.OnCompleted();

    // the completed flag should
    // be reset to True
    Assert.True(completed);
}

Note that our resulting observable stream does the same filtering and transformation as in the previous sample - it filters in only even number and than uses their squares for the output.

The source stream (ISubject<T> subject variable) is populated by calling its OnNext(int i) method and passing some integer to it:

// push data into the stream
// by using OnNext() method
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);

...

// add another even number
subject.OnNext(10);

Then the source stream is completed by calling

// signal source stream completion
subject.OnCompleted();

which also causes the result stream to complete.

The other test TestStreamFromSubjectWithException() from the same file does everything the same as the first test, only signals an error at the end instead of completion:

// signal error
subject.OnError(new Exception("Error Occurred"));

// verify that our onError action worked
Assert.True(errorMessage == "Error Occurred");

The last Assert call verifies that the error message was really passed to Subscribe(...) method's onError argument.

There are several useful variations of Subjects but since my purpose is to avoid writing a handbook, I'll skip them at this point.

C# Events to Streams

There is a close affinity between C# events and Observable Streams. Both implement observable pattern.

In one (not very important) way C# events are more powerful than streams - because they can take an arbitrary number of arguments while Rx stream takes only one (the no argument stream can also be imitated by IObservable<Unit>). This shortcoming can be easily circumvented by using C# Tuples and transforming them with Select(...) method.

There is one very important way in which Rx Observable Streams are much more powerful than C# events - streams can be transformed, manipulated and combined using many LINQ-like operators operators, while the events cannot. Composing events can be a big pain if you ever tried it. Complex event compositions are simply impossible.

StreamFromEvent.cs file has tests demonstrating how to convert an event into an Rx Observable stream.

There is a tiny EventContainingClass class defined. Its only purpose is to contain an Action<bool> event and provide a method for firing it:

class EventContainingClass
{
    // boolean event
    public event Action<bool>? TestEvent;

    // fire event
    public void FireTestEvent(bool arg)
    {
        TestEvent?.Invoke(arg);
    }
}

There is also an extension method GetObservableFromEventObj(this EventContainingClass eventObj) that given an object of that type, returns an Rx observable stream observing TestEvent:

// extension method that 
// gets the observable for the TestEvent
// of the EventContainingClass passed as
// the argument.
private static IObservable<bool> 
    GetObservableFromEventObj(this EventContainingClass eventObj)
{
    // method Observable.FromEvent(...)
    // is used to create an observable 
    // from an event. 
    // The first two arguments to it
    // specify how to add and remove
    // an event handler. 
    IObservable<bool> observable =
        Observable
            .FromEvent<Action<bool>, bool>
            (
            // add handler
            eventHandler =>
                eventObj.TestEvent += eventHandler,

            // remove handler
            eventHandler =>
                eventObj.TestEvent -= eventHandler
            );

    return observable;
}

This method is provided for the sake of the code reuse and it is called multiple times from the tests within the file. It also demonstrates turning an event into an observable stream by using Observable.FromEvent(...) method and providing two arguments to it - first argument specifying how to add a handler to the event and the second how to remove the handler from the event.

The first test method TestStreamFromEvent() simply tests the observable obtained from an event:

[Fact]
public static void TestStreamFromEvent()
{
    // create an object containing an
    // Action<bool> event
    EventContainingClass eventContainingObj = 
        new EventContainingClass();

    // create an observable from 
    // eventContainingObj.TestEvent
    IObservable<bool> observable =
        eventContainingObj.GetObservableFromEventObj();

    // boolean to specify 
    // the last argument to the event
    bool lastFired = false;

    // subscribe to change the lastFired
    // to the last argument passed to 
    // fire the event
    using var subscriptionDisposable =
        observable.Subscribe(b => lastFired = b);

    // false originally
    Assert.False(lastFired);

    // after true is fired, lastFired 
    // should become true
    eventContainingObj.FireTestEvent(true);
    Assert.True(lastFired);

    // after another true is fired, lastFired 
    // should still be true
    eventContainingObj.FireTestEvent(true);
    Assert.True(lastFired);

    // after false is fired, lastFired 
    // should change to false
    eventContainingObj.FireTestEvent(false);
    Assert.False(lastFired);
}

The second test method TestStreamFromEventComposition() demonstrates combining the latest values of two different streams obtained form events by using Observable.CombineLatest(...) method:

[Fact]
public static void TestStreamFromEventComposition()
{
    // first object containing an event
    EventContainingClass eventContainingObj1 =
        new EventContainingClass();

    // observable obtained from the
    // first objects
    IObservable<bool> observable1 =
        eventContainingObj1.GetObservableFromEventObj();

    // second object containing the event
    EventContainingClass eventContainingObj2 =
        new EventContainingClass();

    // observable obtained from the
    // second object
    IObservable<bool> observable2 =
        eventContainingObj2.GetObservableFromEventObj();

    // combine the two observables by using 
    // Observable.CombineLatest(...) extension 
    // method. This method fires every time any 
    // observable changes. 
    IObservable<bool> combinedLatestObservable =
        observable1
            .CombineLatest
            (
                observable2,

                // obtain the combined
                // observable value by 
                // doing boolean AND 
                // operator on the two
                // latest values from the
                // two observables
                 resultSelector: (b1, b2) => b1 && b2
            );

    // get an observable to fire only if the
    // result changes from the previous value
    // (do not emit data if it stays the same)
    IObservable<bool> combinedLatestDistinctObservable =
        combinedLatestObservable.DistinctUntilChanged();

    // latest value from 
    // the resulting observable
    bool latestCompoundValue = false;

    // subscribe to set the
    // latestCompoundValue when 
    // the result changes
    using var subscriptionDisposable =
        combinedLatestDistinctObservable
            .Subscribe
            (
                b => latestCompoundValue = b
            );

    Assert.False(latestCompoundValue);

    eventContainingObj1.FireTestEvent(true);
    eventContainingObj2.FireTestEvent(true);

    // since both observable fired true
    // the result should change to true
    Assert.True(latestCompoundValue);

    // one fire false - result
    // changes to false
    eventContainingObj1.FireTestEvent(false);
    Assert.False(latestCompoundValue);

    // both are false, result 
    // stays false
    eventContainingObj2.FireTestEvent(false);
    Assert.False(latestCompoundValue);

    // only one changed back to true
    // the result still stays false
    eventContainingObj1.FireTestEvent(true);
    Assert.False(latestCompoundValue);

    // both changed to true
    // the result changes to true
    eventContainingObj2.FireTestEvent(true);
    Assert.True(latestCompoundValue);
}

The most interesting and new Rx operators demonstrated here are CombineLatest(...) and DistinctUntilChanged(...).

Observable.CombineLatest(...) combines two or more observable streams by emitting new data when any of its source streams changes (unlike Zip(...) operator that fires when all of its source stream change). Its argument resultSelector specifies how to combine the result from the latest values of the two observables:

// combine the two observables by using 
// Observable.CombineLatest(...) extension 
// method. This method fires every time any 
// observable changes. 
IObservable<bool> combinedLatestObservable =
    observable1
        .CombineLatest
        (
            observable2,

            // obtain the combined
            // observable value by 
            // doing boolean AND 
            // operator on the two
            // latest values from the
            // two observables
            resultSelector: (b1, b2) => b1 && b2
        );

In our case, we use boolean AND operator (&&) to combine the two bool values into a bool result.

The other very interesting method is Observable.DistinctUntilChanged(...). It prevents emitting data unless the data changes (prevents unnecessary stream data emissions).

C# Notifiable Properties to Streams

A property which fires an event every time it changes is very similar to an observable stream. Each property change corresponds to data emission and the data emitted will be just the property value.

And again as in case of events, converting properties to Rx Observable streams enables using the powerful stream manipulation, transform and composition operators.

In Avalonia, WPF and other XAML based frameworks, in order for a view model (non-visual) property change to be picked up by the visuals (the view) the class containing the property has to implement System.ComponentModel.INotifiablePropertyChange interface and the property has to fire PropertyChanged event when its value changes.

The best library for converting notifiable properties to streams and vice versa is ReactiveUI - a platform independent library that references both System.Reactive and DynamicData libraries. RectiveUI is beyond the scope of this article (though later I plan to write about it as part of these series).

DynamicData also has a means of converting a notifiable property to an Rx Stream - static extension method NotifyPropertyChangedEx.WhenValueChanged(...) and we shall demo it in this subsection.

The test is located within StreamFromNotifiableProperty.cs file. The file also contains an PropContainer class which has NotifiableProperty of type bool in it that fires PropertyChanged(...) event when the property changes;

// class with single notifiable property 
class PropContainer : INotifyPropertyChanged
{
    public event PropertyChangedEventHandler? PropertyChanged;

    bool _notifiableProperty = false;
    public bool NotifiableProperty
    {
        get => _notifiableProperty;

        set
        {
            if (_notifiableProperty == value)
            {
                return;
            }

            _notifiableProperty = value;

            // fire PropertyChanged event 
            // if the property changes.
            PropertyChanged?.Invoke
            (
                this, 
                new PropertyChangedEventArgs(nameof(NotifiableProperty))
            );
        }
    }
}

And here is the code of the test:

[Fact]
public static void TestNotifyablePropertyValueChange ()
{
    // create an object containing a notifiable property
    var propContainer = new PropContainer ();

    // use DynamicData's 
    // NotifyPropertyChangedEx.WhenValueChanged(...)
    // method to convert the property to IObservable<bool>
    IObservable<bool> whenPropChangedObservable = 
        propContainer.WhenValueChanged
        (
            pContainer => pContainer.NotifiableProperty
        );

    var currentPropValue = propContainer.NotifiableProperty;

    // subscribe to change currentPropValue
    // every time the property changes via the observable
    using var subscriptionDisposable = 
        whenPropChangedObservable
            .Subscribe(newValue => currentPropValue = newValue);

    // test that the currentPropValue (obtained through 
    // the observable) really matches the current property.
    Assert.Equal
    (
        currentPropValue, 
        propContainer.NotifiableProperty
    );

    propContainer.NotifiableProperty = true;
    Assert.Equal
    (
        currentPropValue, 
        propContainer.NotifiableProperty
    );

    propContainer.NotifiableProperty = false;
    Assert.Equal
    (
        currentPropValue, 
        propContainer.NotifiableProperty
    );
}

The way we get the property observable is by calling WhenValueChanged(...) operator on the object containing the property and passing an C# expression corresponding to that property to it:

// use DynamicData's 
// NotifyPropertyChangedEx.WhenValueChanged(...)
// method to convert the property to IObservable<bool>
IObservable<bool> whenPropChangedObservable = 
    propContainer.WhenValueChanged
    (
        pContainer => pContainer.NotifiablePropert
    );

Note that DynamicData also contains some other useful property notification converter operators including

WhebPropertyChanged(...) - creates an observable that returns the changed object also (on top of the current property value). This can be important e.g. when dealing with collections.
WhenChanged(...) methods allowing to monitor several properties with single observable.

Dynamic Data Collections To Rx Observable Streams (Preview)

I plan to start talking about DynamidData in-depth further in the next article of the series. So here i'll just give a bit of a pep talk about the main ideas behind Dynamic Data.

C# has a built-in ObservableCollection<T> class which implements INotifyCollectionChanged interface and fires CollectionChanged event each time the items are added, removed or rearranged within the collection.

The arguments to CollectionChanged event carry full information about the changed items sufficient to reconstruct the changed collection.

Potentially CollectionChanged arguments can be used as an IObservable<T> streams allowing to use the collection changes to create and update other collections dependent on the original one.

Instead of ObservableCollection<T>, the Dynamic Data uses ISourceList<T> and ISourceCache<T, TKey> as collection containers.

ISourceCache<T, TKey> assumes that the collection items have a primary key and based on the primary key of incoming data - we either insert a new item or update the current item (with the same primary key) within the collection.

Both ISourceList<T> and ISourceCache<T, TKey> have a Connect() method that allows to get IObservable<IChangeSet<T>> (in case of ISourceList) and IObservable<IChangeSet<T, TKey>> (in case of the ISourceCache). These observables are nothing more than the observables of changes to the corresponding collections. Note that IChangeSet is not identical to CollectionChanged event arguments, but is optimized for better performance and collection reconstruction.

Using IObservable<IChangeSet<T>> and IObservable<IChangeSet<T, TKey>> we can create other collections that mimic the original one, with some difference provided by the operations performed on IObservable. For example we can create a collection that filters in only even int number from the original collection of integers (or something much much more complex).

The main trick is that such dependencies between the collections will be maintained automagically by Dynamic Data even if the source collection(s) change.

More about Dynamic data is coming in the future articles.

Nick Polyak's Software Articles are Coming to Dev.To

Nick Polyak — Thu, 23 Jan 2025 23:33:23 +0000

For many years (more that a decade) codeproject.com used to be my home for publishing software architecture and development related articles.
Now since codeproject is unfortunately unavailable (hopefully only temporarily) I plan to make Dev.To to be my software blog home possibly with mirrors at other software blog hosting web sites.

DEV Community: Nick Polyak

Introduction to Dynamic Data's Observable Cache

Intro to Dynamic Data

Source Code Location

Simplified Trade class

Basic SourceCache<T, TKey> API

Creating SourceCache<Trade, int>

Adding or Updating a Trade to Cache

Using IObservableCache<T, TKey>

Turning SourceCache<T, TKey> (or IObservableView<T, TKey>) into an Rx Observable Stream of Changes to the Collection using Connect() Method

Turning ObservableCollection<T> into an Rx Observable Stream of Collection Changes

Code Samples

Auxiliary Methods and APIs used throughout the Tests

Filtering, Sorting and Binding Test

Filtering, Sorting and Binding Test with ObservableCollection<Trade> as Source.

Dynamic Filter Test

Grouping Test

Introduction to Dynamic Data

What is DynamicData for?

Source Code Location

Introductory Sample

Dynamic Filter Demo

Total Sum Aggregation Tests

Grouping Sample

Grouping with Binding and Aggregation Sample

[Boost]

Introduction to Rx Observable Streams and Dynamic Data

Nick Polyak ・ Feb 7

[Boost]

Using XUnit Test Framework for Testing C# Code under Visual Studio

Nick Polyak ・ Feb 9

Everything you need to know to create XUnit C# Tests in Visual Studio

Using Rx Observable Streams and Dynamic Data Patterns to build Applications Clean and Fast (Introduction)

What this article is about

Introduction to Reactive Extensions (Rx)

Introduction to Dynamic Data

Rx.NET Streams

Reactive Stream Sources

Using Streams of Remote Data

Using System.Reactive and DynamicData Functionality to Create Streams

Converting Collection into IObservable<T>

IAsyncEnumerable to Stream

Using Subjects for Creating Observable Streams.

C# Events to Streams

C# Notifiable Properties to Streams

Dynamic Data Collections To Rx Observable Streams (Preview)

Nick Polyak's Software Articles are Coming to Dev.To

Simplified `Trade` class

Basic `SourceCache<T, TKey>` API

Creating `SourceCache<Trade, int>`

Using `IObservableCache<T, TKey>`

Turning `SourceCache<T, TKey>` (or `IObservableView<T, TKey>`) into an Rx Observable Stream of Changes to the Collection using `Connect()` Method

Turning `ObservableCollection<T>` into an Rx Observable Stream of Collection Changes

Filtering, Sorting and Binding Test with `ObservableCollection<Trade>` as Source.

Converting Collection into `IObservable<T>`