Creating a Reusable ‘ObservableSource’ for the Reactive Extensions (part 5a)

Updates for Next Reactive Extensions (Rx) Code Drop

The new code drop came through, 2½ months after the last drop, on 5^th March. The release notes are suitably long. With a fair amount of change, much of it under the covers to edge cases.

But a fair number of things have moved assemblies and changed name.

Impact on My Implementation

Given all the change, there was relatively little impact. The only thing that broke was my use of Notification<T>: the Current property is now called Value. And it now has an Exception property, which saves a cast when handling an OnError notification.

My (unpublished) test harness used GroupDisposable to clean up subscriptions, this was renamed CompositeDisposable and moved to the CoreEx assembly.

Both of these renames give slightly better and more consistent names, and the addition of the Exception property is definitely better. So overall the new drop is, for me, an improvement so far: once the code compiled all tests just passed.

The only function that changed was SendNotification so following DRY helped. The new implementation:

private static void SendNotification(IObserver&lt;T&gt; obs, Notification&lt;T&gt; n) {
    switch (n.Kind) {
    case NotificationKind.OnNext:
        Debug.Assert(n.HasValue);
        obs.OnNext(n.Value);
        break;
    case NotificationKind.OnCompleted:
        obs.OnCompleted();
        break;
    case NotificationKind.OnError:
        obs.OnError(n.Exception);
        break;
    }
}

So Where’s The Concurrency Support…?

Coming! :-)
I've been rather head down in learning WPF for something else, and trying to avoid too many distractions while getting my head around the basics of WPF. Concurrency support is perhaps the biggest challenge here and I therefore want to be able to focus on it and get it right. This will mean taking a block of focused time.